spring XD结合RabbitMQ方法介绍

zgdy

1，Rabbitmq工作流程
　　消息生产者（即下图中的ClientA，ClientB）生产message并交给交换机Exchange，Exchange将消息按照RoutingKey分发到相应的队列Queue中，而我们的消费者（即下图中的Client1，Client2，Client3）需要做的事情就是监听队列Queue的事件，当有新的message到达的时候，做出相应的处理。

Rabbitmq结构图；

2，Exchange交换类型介绍
　　具体来说，Exchange也有几个类型，完全根据key进行投递的叫做Direct交换机，例如，绑定时设置了routing key为“abc”，那么客户端提交的消息，只有设置了key为“abc”的才会投递到队列。对key进行模式匹配后进行投递的叫做Topic交换机，符号“#”匹配一个或多个词，符号“*”匹配正好一个词。例如“abc.#”匹配“abc.def.ghi”，“abc.*”只匹配“abc.def”。还有一种不需要key的，叫做“fanout”交换机，它采取广播模式，一个消息进来时，投递到与该交换机绑定的所有队列；我们应该使用的就是“fanout”交换机。
3，“注册”新消费者的流程
　　在我们“注册”新的消费者的时候，首先要做的事情就是明确连接的host地址（ip）和Exchange的name，然后声明queue的时候将其绑定到对应name的Exchange上即可。
4，spring XD与Rabbitmq的结合方法
　　在xd-shell中执行stream create --name RABBITMQ_NAME --definition "http | rabbit --exchange= EXCHANGE_NAME" –deploy
　　其中，替换RABBITMQ_NAME为rabbitmq的名字，EXCHANGE_NAME替换为Exchange的名字，并与消费者绑定的Exchange的名字保持一致；
5,测试方法：
　　curl -X POST -d 'helloworld！！！' http://host_ip:host_port
　　将host_ip和host_post设置为Rabbitmq所在主机ip和端口，启动消费者程序，如果能够收到message “helloworld！！！”则可证明成功！
6，消费者示例代码：
　　package main.java.rabbitmq;
　　import java.io.IOException;
　　import java.util.concurrent.TimeoutException;
　　import com.rabbitmq.client.*;
　　public class Recv {
　　private static final String EXCHANGE_NAME = "bot";
　　public static void main(String[] argv) throws Exception {
　　//创建连接连接到Rabbitmq
　　ConnectionFactory factory = new ConnectionFactory();
　　//设置Rabbitmq所在主机ip或者主机名
　　factory.setHost("localhost");
　　//创建一个连接；
　　Connection connection = factory.newConnection();
　　//创建一个频道
　　Channel channel = connection.createChannel();
　　//频道的exchange声明
　　channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
　　//获取队列名称
　　String queueName = channel.queueDeclare().getQueue();
　　System.out.println("queueName [" + queueName + "]");
　　//绑定队列与exchange
　　channel.queueBind(queueName, EXCHANGE_NAME, "");
　　System.out.println("

Waiting for messages. To exit press CTRL+C");
　　//配置好获取消息的方式
　　QueueingConsumer consumer = new QueueingConsumer(channel);
　　channel.basicConsume(queueName, true, consumer);
　　while(true){
　　//获取消息，如果没有消息，这一步将会一直阻塞
　　QueueingConsumer.Delivery delivery = consumer.nextDelivery();
　　String message = new String(delivery.getBody());
　　System.out.println("Received [" + message + "]"); }
　　}
　　}