RabbitMQ-fanout广播模式

迷南。 2022-02-26 02:22 308阅读 0赞

### RabbitMQ-fanout广播模式 ###

fanout: 所有bind到该交换机的队列都可以接收消息。

消息的接受者需要创建一个队列，绑定到该fanout交换机上，监听来自发送端的消息广播。fanout模式和topic模式基本是一样的，唯一的一个区别在于topic可以根据不同的主题路由到不同的队列，但是fanout模式是没有路由这个过程的，所有绑定到该交换机上的队列都会收到这一条广播消息。  
![在这里插入图片描述][20190328181306195.png]  
     基于上文中封装的RabbitMq代码[RabbitMQ C++封装][RabbitMQ C]，在此实现fanout广播消息的发送和消息。  
     发送端代码：

std::string err;
    
    	//mq连接
    	CRabbitMqClient mqClient;
    	if (0 != mqClient.Connect(err))
    	{
    		return -1;
    	}
    
    	//声明交换机
    	CExchange exchange(exchange_name, "fanout", true, false);
    	if (0 != mqClient.DeclareExchange(exchange, err))
    	{
    		return -1;
    	}
    
    	amqp_basic_properties_t properties;
    	GetProperties(properties);
    	for (int i = 0; i < 5; i++)
    	{
    		std::string szBuf = "test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1test-1";
    		CMessage message(szBuf, properties, routing_keys_name);
    		if (0 != mqClient.PublishMessage(message, err))
    		{
    			return -1;
    		}
    		//printf("发送消息成功：%s\n", szBuf);
    	}

消费端代码，需要找到对应的交换机，并创建属于自己的队列，绑定到该交换机上，进行广播消息的接收：

std::string err;
    
    	//mq连接
    	CRabbitMqClient mqClient;
    	if (0 != mqClient.Connect(err))
    	{
    		return -1;
    	}
    
    	//声明交换机
    	CExchange exchange(exchange_name, "fanout", true, false);
    	if (0 != mqClient.DeclareExchange(exchange, err))
    	{
    		return -1;
    	}
    
    	//声明队列
    	CQueue queue(queue_name);
    	if (0 != mqClient.DeclareQueue(queue, err))
    	{
    		return -1;
    	}
    
    	//队列绑定到交换机上
    	if (0 != mqClient.BindQueueToExchange(queue, exchange, routing_keys_name, err))
    	{
    		return -1;
    	}
    
    	do 
    	{
    		std::vector<std::string> messageVec;
    		mqClient.ConsumerMessage(messageVec, queue_name);
    		printf("接收消息成功：\n");
    		for (int i = 0; i < messageVec.size(); i++)
    		{
    			printf("消息%d：%s\n", i, messageVec[i].c_str());
    		}
    	} while (1);

消费端需要声明一个和生产端一模一样的交换机（包括交换机的名称，属性），然后再声明一个临时队列（各个消费者需要各自声明一个唯一队列，因为RabbitMq只会把消息发送到一个队列一次，比如：comsumer1和comsumer2声明了同一个队列，producer生产了3条消息，那么可能comsumer1收到了2条，comsumer2收到了1条；也有可能comsumer1收到了3条，comsumer2一条都没收到），绑定到该交互机上，在一个while循环中不停消费该队列上的消息。

[20190328181306195.png]: /images/20220226/4a25cffc4d4c4944b0ab93926c2a5096.png
[RabbitMQ C]: https://blog.csdn.net/bajianxiaofendui/article/details/88875446