RabbitMQ协议基础及C++和Java混合开发

#include
#include "event2/event.h"
#include "amqpcpp.h"
#include "amqpcpp/libevent.h"
#include "amqp_msg.pb.h"
#include <string>
using namespace std;
using namespace amqp;
int main()
{
event_base *base = event_base_new(); // 通过libevent启动实践循环
AMQP::LibEventHandler handler(base);
AMQP::TcpConnection connection(&handler, AMQP::Address("localhost", 5672, AMQP::Login("guest", "guest"), "/"));
AMQP::TcpChannel channel(&connection); // 创建一条通道
channel.setQos(1);
// 监听login.rpc队列
channel.consume("login.rpc").onReceived([&](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered)
{
cout <<"login.rpc" << endl;
Login login;
login.ParseFromArray(message.body(), message.bodySize());
Response resp; // 创建应答对象
resp.set_status(Response_RespType_OK);
resp.set_session_id("acd");
char data[1024] = {0};
int data_size = resp.ByteSizeLong();
resp.SerializeToArray(data, data_size);
AMQP::Envelope env(data, data_size);
env.setCorrelationID(message.correlationID()); // 获取应答ID
channel.publish("", message.replyTo(), env); // 发送给应答队列
channel.ack(deliveryTag); // 向MQ发送确认
}).onSuccess([&](const std::string &consumertag)
{
}).onError([](const char *message)
{
event_base_loopbreak(base); // 发送错误中断事件循环
cout < endl;
});
// 监听logout.rpc队列
channel.consume("logout.rpc")
.onReceived([&channel](const AMQP::Message &message, uint64_t deliveryTag, bool)
{
Logout logout;
logout.ParseFromArray(message.body(), message.bodySize());
Response resp;
resp.set_status(Response_RespType_OK);
char data[1024] = {0};
int data_size = resp.ByteSizeLong();
resp.SerializeToArray(data, data_size);
AMQP::Envelope env(data, data_size);
env.setCorrelationID(message.correlationID());
channel.publish("", message.replyTo(), env);
channel.ack(deliveryTag);
}).onError([](const char *message)
{
event_base_loopbreak(base);
cout < endl;
});
event_base_dispatch(base); // 事件循环
event_base_free(base); // 释放
return 0;
}

channel.onError([&base](const char* message)
{
std::cout <<"Channel error: " < std::endl;
event_base_loopbreak(base);
});

channel.declareQueue("queueName", AMQP::passive)
.onSuccess([&](const string& name, uint32_t, uint32_t)
{
cout <<"Queue Name:" < endl;
});

channel.declareExchange("logs", AMQP::fanout)
.onSuccess([&]() {})

@GetMapping("login")
public String loginRpc() throws InvalidProtocolBufferException {
AmqpMsg.Login login = AmqpMsg.Login.newBuilder()
.addParams(AmqpMsg.PairParams.newBuilder().setKey("username").setValue("admin").build())
.addParams(AmqpMsg.PairParams.newBuilder().setKey("password").setValue("admin").build())
.build();
byte[] resp = (byte[]) template.convertSendAndReceive(directExchange.getName(), "login.rpc", login.toByteArray());
AmqpMsg.Response response = AmqpMsg.Response.parseFrom(resp);
if (response.getStatus() == AmqpMsg.Response.RespType.OK) {
String sessionID = response.getSessionId();
System.out.println("登录成功 SessiOnID=" + sessionID);
return "SUCCESS";
}
return "ERROR";
}
@GetMapping("logout")
public String logoutRpc() throws InvalidProtocolBufferException {
AmqpMsg.Logout logout = AmqpMsg.Logout.newBuilder()
.setSessionId("123456").build();
byte[] resp = (byte[]) template.convertSendAndReceive(directExchange.getName(), "logout.rpc", logout.toByteArray());
AmqpMsg.Response response = AmqpMsg.Response.parseFrom(resp);
if(response.getStatus() == AmqpMsg.Response.RespType.OK) {
System.out.println("注销成功");
return "SUCCESS";
}
return "ERROR";
}

@Configuration
public class RPCRabbitConfig {
@Bean("simple")
public Queue simpleQueue() {
return new Queue("simple");
}
@Bean("login.rpc")
public Queue loginRpcQueue() {
return new Queue("login.rpc");
}
@Bean("logout.rpc")
public Queue logoutRpcQueue() {
return new Queue("logout.rpc");
}
@Bean
public DirectExchange defaultExchange() {
return new DirectExchange("amq.direct");
}
@Bean
public Binding loginRpcBinding(DirectExchange exchange, @Qualifier("login.rpc") Queue queue) {
return BindingBuilder.bind(queue).to(exchange).with("login.rpc");
}
@Bean
public Binding logoutRpcBind(DirectExchange exchange, @Qualifier("logout.rpc") Queue queue) {
return BindingBuilder.bind(queue).to(exchange).with("logout.rpc");
}
}