rabbitMq 初步
rabbitmq的工作原理
它的基本结构
组成部分说明如下：
broker：消息队列服务进程，此进程包括两个部分：exchange和queue。
exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。
producer：消息生产者，即生产方客户端，生产方客户端将消息发送到mq。
consumer：消息消费者，即消费方客户端，接收mq转发的消息。
maven举例配置
<dependency> <groupid>com.rabbitmq</groupid> <artifactid>amqp‐client</artifactid> <version>4.0.3</version><!‐‐此版本与spring boot 1.5.9版本匹配‐‐> </dependency> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring‐boot‐starter‐logging</artifactid> </dependency>
生产者举例demo
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.qualifier;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
@configuration
public class rabbitmqconfig {
public static final string queue_inform_email = "queue_inform_email";
public static final string queue_inform_sms = "queue_inform_sms";
public static final string exchange_topics_inform="exchange_topics_inform";
public static final string routingkey_email="inform.#.email.#";
public static final string routingkey_sms="inform.#.sms.#";
//声明交换机
@bean(exchange_topics_inform)
public exchange exchange_topics_inform(){
//durable(true) 持久化，mq重启之后交换机还在
return exchangebuilder.topicexchange(exchange_topics_inform).durable(true).build();
}
//声明queue_inform_email队列
@bean(queue_inform_email)
public queue queue_inform_email(){
return new queue(queue_inform_email);
}
//声明queue_inform_sms队列
@bean(queue_inform_sms)
public queue queue_inform_sms(){
return new queue(queue_inform_sms);
}
//routingkey_email队列绑定交换机，指定routingkey
@bean
public binding binding_queue_inform_email(@qualifier(queue_inform_email) queue queue,
@qualifier(exchange_topics_inform) exchange exchange){
return bindingbuilder.bind(queue).to(exchange).with(routingkey_email).noargs();
}
//routingkey_sms队列绑定交换机，指定routingkey
@bean
public binding binding_routingkey_sms(@qualifier(queue_inform_sms) queue queue,
@qualifier(exchange_topics_inform) exchange exchange){
return bindingbuilder.bind(queue).to(exchange).with(routingkey_sms).noargs();
}
}
消费者举例demo
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.qualifier;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
@configuration
public class rabbitmqconfig {
public static final string queue_inform_email = "queue_inform_email";
public static final string queue_inform_sms = "queue_inform_sms";
public static final string exchange_topics_inform="exchange_topics_inform";
public static final string routingkey_email="inform.#.email.#";
public static final string routingkey_sms="inform.#.sms.#";
//声明交换机
@bean(exchange_topics_inform)
public exchange exchange_topics_inform(){
//durable(true) 持久化，mq重启之后交换机还在
return exchangebuilder.topicexchange(exchange_topics_inform).durable(true).build();
}
//声明queue_inform_email队列
@bean(queue_inform_email)
public queue queue_inform_email(){
return new queue(queue_inform_email);
}
//声明queue_inform_sms队列
@bean(queue_inform_sms)
public queue queue_inform_sms(){
return new queue(queue_inform_sms);
}
//routingkey_email队列绑定交换机，指定routingkey
@bean
public binding binding_queue_inform_email(@qualifier(queue_inform_email) queue queue,
@qualifier(exchange_topics_inform) exchange exchange){
return bindingbuilder.bind(queue).to(exchange).with(routingkey_email).noargs();
}
//routingkey_sms队列绑定交换机，指定routingkey
@bean
public binding binding_routingkey_sms(@qualifier(queue_inform_sms) queue queue,
@qualifier(exchange_topics_inform) exchange exchange){
return bindingbuilder.bind(queue).to(exchange).with(routingkey_sms).noargs();
}
}
工作模式
rabbitmq有以下几种工作模式 ：
1、work queues
2、publish/subscribe
3、routing
4、topics
5、header
6、rpc
work queues
work queues与入门程序相比，多了一个消费端，两个消费端共同消费同一个队列中的消息。
应用场景：对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。
测试：
1、使用入门程序，启动多个消费者。
2、生产者发送多个消息。
结果：
1、一条消息只会被一个消费者接收；
2、rabbit采用轮询的方式将消息是平均发送给消费者的；
3、消费者在处理完某条消息后，才会收到下一条消息。
publish/subscribe 发布订阅模式
发布订阅模式：
1、每个消费者监听自己的队列。
2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收
到消息
routin
路由模式：
1、每个消费者监听自己的队列，并且设置routingkey。
2、生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列。
这是一种非常灵活的模式，经常被用到
topics
路由模式：
1、每个消费者监听自己的队列，并且设置带统配符的routingkey。
2、生产者将消息发给broker，由交换机根据routingkey来转发消息到指定的队列。
header模式
header模式与routing不同的地方在于，header模式取消routingkey，使用header中的 key/value（键值对）匹配
队列。
案例：
根据用户的通知设置去通知用户，设置接收email的用户只接收email，设置接收sms的用户只接收sms，设置两种
通知类型都接收的则两种通知都有效。
生产者demo：
map<string, object> headers_email = new hashtable<string, object>();
headers_email.put("inform_type", "email");
map<string, object> headers_sms = new hashtable<string, object>();
headers_sms.put("inform_type", "sms");
channel.queuebind(queue_inform_email,exchange_headers_inform,"",headers_email);
channel.queuebind(queue_inform_sms,exchange_headers_inform,"",headers_sms);
通知demo :
string message = "email inform to user"+i;
map<string,object> headers = new hashtable<string, object>();
headers.put("inform_type", "email");//匹配email通知消费者绑定的header
//headers.put("inform_type", "sms");//匹配sms通知消费者绑定的header
amqp.basicproperties.builder properties = new amqp.basicproperties.builder();
properties.headers(headers);
//email通知
channel.basicpublish(exchange_headers_inform, "", properties.build(), message.getbytes());
发送邮件消费者 :
channel.exchangedeclare(exchange_headers_inform, builtinexchangetype.headers);
map<string, object> headers_email = new hashtable<string, object>();
headers_email.put("inform_email", "email");
//交换机和队列绑定
channel.queuebind(queue_inform_email,exchange_headers_inform,"",headers_email);
//指定消费队列
channel.basicconsume(queue_inform_email, true, consumer);
rpc
rpc即客户端远程调用服务端的方法 ，使用mq可以实现rpc的异步调用，基于direct交换机实现，流程如下：
1、客户端即是生产者就是消费者，向rpc请求队列发送rpc调用消息，同时监听rpc响应队列。
2、服务端监听rpc请求队列的消息，收到消息后执行服务端的方法，得到方法返回的结果
3、服务端将rpc方法 的结果发送到rpc响应队列
4、客户端（rpc调用方）监听rpc响应队列，接收到rpc调用结果。