AMQP-cpp 开发

2019-11-08 03:22:50

字体：大中小

来源：转载

供稿：网友

linux开发：从官网下载amqp-cpp（https://github.com/CopernicaMarketingSoftware/AMQP-CPP）这个库，这个库是linux 下的。里面包含了linux下的tcp通信（不兼容widows），适合linux开发。

windows开发：由于官网没有提供可以直接在windows下使用的amqp-cpp，所以需要将官网的linux下的库经过修改，剥离了tcp通信部分，只留下了AMQP协议框架部分（https://git.oschina.net/Ailsc/amqp-cpp.git）。所以开发的话需要自己加入tcp（iocp：https://git.oschina.net/Ailsc/jeflib.git）通信部分配合AMQP-cpp协议框架，来共同开发AMQP。amqp-cpp.git内部有完整的Demo。

AMQP开发之前需要了解AMQP协议，可以参考前面几章对vhost，exchange，que的参数做一个基本了解。然后配合AMQP-cpp进行软件开发。

开发流程： 1.实现自己的AMQP::ConnectionHandler，ConnectionHandler类似于一个中间组件，将AMQP请求指令转化为数据流，然后通过Tcp将数据流发送出去。

class CTcpConnectionHandle : public AMQP::ConnectionHandler{public: CTcpConnectionHandle(CTcpLink *plink); virtual ~CTcpConnectionHandle();public: ///> 当AMQP发送数据的时候触发onData virtual void onData(AMQP::Connection *connection, const char *buffer, size_t size); ///> 当AMQP协议准备好（已经登录成功）触发 virtual void onConnected(AMQP::Connection *connection); ///> 这个需要在tcp 断开的时候手动调用一下 virtual void onClosed(AMQP::Connection *connection); ///> 当发生错误的时候，该连接不可再用需要重新建立连接 virtual void onError(AMQP::Connection *connection, const char *message);PRivate: CTcpLink *m_plink;//和当前连接关联};

2.实现TcpLink进行数据流的收发操作，以及数据解析。

class CTcpLink :public jeflib::iocp::ILinkContext{ friend class CTcpConnectionHandle;public: CTcpLink(); ~CTcpLink();public: ///> set usr psw vhost void setAMQP(std::string strusr, std::string strpsw, std::string strvhost); ///> block bool IsReady(); virtual void on_close();//on connection close,主动断开不会收到通知 ///> for tcp svr call back virtual void on_accepted(jeflib::iocp::NETHANDLE nethandle, const char *szip, const unsigned short sport){}; ///> tcp recv data virtual void on_recv(const char *pdata, int ndatasize); ///> client connect ok virtual void on_connect_ok(jeflib::iocp::NETHANDLE nethandle); ///> client connect error virtual void on_connect_err(jeflib::iocp::NETHANDLE nethandle, bool bactive/*是否为主本端动关闭*/); ///> for iocp send data call back to resend virtual void on_send_ok(const char *pdata, int ndatasize){} virtual void on_send_err(const char *pdata, int ndatasize){} virtual CTcpLink* new_context(){ return new CTcpLink; } bool send(const char *data, int size); void parse(); Operator AMQP::Connection&(){ return *m_pConnect;}protected: std::promise<bool> m_bready;//是否准备好 bool m_blink; mutable std::mutex m_lock; std::vector<char> m_recv_buff;//数据接收缓冲区 AMQP::Connection *m_pConnect;//AMQP connect CTcpConnectionHandle *m_phandler; bool m_bpares;private://AMQP 登录信息 std::string m_strusr; std::string m_strpsw; std::string m_strvhost;};

3.AMQP::Connection 这个是将网络来的数据解析为AMQP协议，并且将数据行为反应到对应的回调函数之中。

void CTcpLink::parse(){ if (m_pConnect == NULL) return; uint64_t use = 0; ///> 不阻塞线程，parse一次只能一个线程调用，不能多个线程 std::unique_lock<std::mutex> guard(m_lock, std::try_to_lock); if (!guard.owns_lock()) return; if (m_bpares) return; m_bpares = true; size_t size = m_recv_buff.size(); while (size - use >= m_pConnect->expected()) { std::vector<char> buff(m_recv_buff.begin() + use, m_recv_buff.begin() + use + m_pConnect->expected()); ///> 解析期间打开锁，允许接收数据 guard.unlock(); use += m_pConnect->parse(buff.data(), buff.size()); guard.lock(); } m_recv_buff.erase(m_recv_buff.begin(), m_recv_buff.begin() + use); m_bpares = false;}

4.连接Rabbit服务器，然后进行AMQP的通信。通信流程

Y-操作成功连接服务器-》（Y）创建通道=》（Y）启用ACK=》（Y）声明exchange=》（Y）声明队列=》（Y）bind 队列=》（Y）开始消费队列,publish队列

异常处理: tcp连接异常:可以采取重连或者关闭连接。 rabbit登录异常:断开Tcp连接。通道异常:关闭当前通道，然后重启一个通道，复制该异常通道的业务。

if(channel ready){ if(declare que) { channel->bind current que； channel->declare next que； } else//声明队列失败 { 从需要声明的队列中移除当前错误的队列，因此即使再次尝试声明该队列还是会异常，所以抛弃该队列。 channel->erase(que); 获取当前所有需要声明的队列 aryque = channel->getqueary(); 重启通道，并且将需要声明的队列让新通道去做 new channel(aryque ); }}

具体AMQP实现请参照上述Demo。

注意事项： 1.队列的Auto delete 属性，只有已经发生过consumer操作的时候才会生效。 2.只有触发AMQP::ConnectionHandler::onConnected才能表明AMQP协议准备完成，包括连接和登录准备完毕，才能进行AMQP操作。 3.AMQP的通道不是线程安全的，在进行通道操作的时候需要加锁。或者一个线程操作一个通道。 4.AMQP同一个通道的指令处理顺序需要保证序列性，因为一个请求或者指令可能存在几个通信包。例如：publis存在三次send才完成一次publish指令。否则可能导致指令交汇，导致异常。

上一篇：关于子父类中调用顺序的小结

下一篇：swift中的基本运算