spring XD结合RabbitMQ方法介绍

Posted 2020-08-15

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 + "]"); }

}

}