首页 > 代码库 > spring XD结合RabbitMQ方法介绍

spring XD结合RabbitMQ方法介绍

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

}

}

spring XD结合RabbitMQ方法介绍