首页 > 代码库 > 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方法介绍