RabbitMQ入门学习笔记（二）工作模式
1、Work queues
2、Publish/Subscribe
3、Routing
4、Topics
5、Header
6、RPC
下面一一介绍
测试：
2、生产者发送多个消息。
结果：
1、一条消息只会被一个消费者接收；
2、rabbit采用轮询的方式将消息是平均发送给消费者的；
3、消费者在处理完某条消息后，才会收到下一条消息
案例：
用户通知，当用户充值成功或转账完成系统通知用户，通知方式有短信、邮件多种方法 。
生产者代码示例
public class Producer02_publish {
//email消息队列
private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
//sms消息队列
private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
//交换机
private static final String EXCHANGE_FANOUT_INFORM = "exchange_fanout_inform";
public static void main(String[] args) {
//连接
Connection connection = null;
//通道
Channel channel = null;
try {
//给MQ发送消息
//连接MQ
//通过连接工厂创建连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");//IP地址
connectionFactory.setPort(5672);//默认mq服务端口
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");//设置mq虚拟机
//和MQ创建连接
connection = connectionFactory.newConnection();
//建立channel通道，会话通道，在通道中向mq发送消息
channel = connection.createChannel();
//声明交换机
/**
* String exchange, BuiltinExchangeType type
* 1、交换机名称,采用发布订阅模式，交换机的类型要设置为fanout，Routing路由模式将交换机的类型设置为direct
* 如果使用的是Topics模式将交换机的类型设置为topics。。使用HEADERS模式将交换机类型设置为HEADERS
*/
channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);
//声明一个队列,根据队列名称判断，如果在mq中没有此队列就创建一个队列，如果有此队列则不作处理
/**
* 参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
* 1、队列名称
* 2、是否持久化，true表示持久化，当mq重启之后此队列还在
* 3、是否独占通道，true表示此通道只能由此队列来使用
* 4、自动删除，true表示自动删除，mq重启后队列删除
* 5、队列参数列表
*/
channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
//将队列绑定到交换机中,目的就是让交换机将消息转发到队列
/**
* 参数：String queue, String exchange, String routingKey
* 1、队列名称
* 2、交换机名称
* 3、routingKey路由key，交换机根据路由key将消息转发到队列,发布订阅模式不需要指定routingkey
*/
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,"");
channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_FANOUT_INFORM,"");
//发送消息
/**
* 参数String exchange, String routingKey, BasicProperties props, byte[] body；
* 1、exchange 交换机名称
* 2、routingKey 对于发布订阅模式不需要指定routingkey
* 3、消息属性
* 4、消息内容
*/
for(int i=0;i<10;i++){
String message ="send message:"+ System.currentTimeMillis();
channel.basicPublish(EXCHANGE_FANOUT_INFORM,"",null,message.getBytes());
System.out.println("send message.."+message);
}
邮件消费者部分代码示例
public class Consumer02_subscribe_email {
//email消息队列
private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
//交换机
private static final String EXCHANGE_FANOUT_INFORM = "exchange_fanout_inform";
public static void main(String[] args) {
//连接
Connection connection = null;
//通道
Channel channel = null;
try {
//给MQ发送消息
//连接MQ
//通过连接工厂创建连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");//IP地址
connectionFactory.setPort(5672);//默认mq服务端口
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");//设置mq虚拟机
//和MQ创建连接
connection = connectionFactory.newConnection();
//建立channel通道，会话通道，在通道中向mq发送消息
channel = connection.createChannel();
//声明交换机
/**
* String exchange, BuiltinExchangeType type
* 1、交换机名称,采用发布订阅模式，交换机的类型要设置为fanout，Routing路由模式将交换机的类型设置为direct
* 如果使用的是Topics模式将交换机的类型设置为topics。。使用HEADERS模式将交换机类型设置为HEADERS
*/
channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);
//声明一个队列,根据队列名称判断，如果在mq中没有此队列就创建一个队列，如果有此队列则不作处理
/**
* 参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
* 1、队列名称
* 2、是否持久化，true表示持久化，当mq重启之后此队列还在
* 3、是否独占通道，true表示此通道只能由此队列来使用
* 4、自动删除，true表示自动删除，mq重启后队列删除
* 5、队列参数列表
*/
channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
//将队列绑定到交换机中,目的就是让交换机将消息转发到队列
/**
* 参数：String queue, String exchange, String routingKey
* 1、队列名称
* 2、交换机名称
* 3、routingKey路由key，交换机根据路由key将消息转发到队列,发布订阅模式不需要指定routingkey
*/
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,"");
//创建回调方法类
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
* 消费者接收到消息后会调用此方法
* @param consumerTag 消费者标签，用来标识消费，如果不指定默认一个名称
* @param envelope 消息内容包
* @param properties 消息的属性
* @param body 消息的内容
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String exchange = envelope.getExchange();//交换
long deliveryTag = envelope.getDeliveryTag();//消息id
String routingKey = envelope.getRoutingKey();//路由key
String message = new String(body, "utf-8");
System.out.println(message);
}
};
//监听队列,接收消息
/**
* 参数：String queue, boolean autoAck, Consumer callback
* 1、监听队列的名称
* 2、autoAck是否自动回复，消息者接收到消息要给mq回复表示此消息已接收，此时mq去删除消息
* 如果autoAck设置true，表示消费者接收到消息后自动回复，如果设置为false，需要程序员在代码中手动回复(channel.basicAck();)
* 3、回调方法，接收到消息后调用此callback
*/
channel.basicConsume(QUEUE_INFORM_EMAIL,true,consumer);
短信消费者代码同上面差不多不再写了
测试
使用生产者发送若干条消息，每条消息都转发到各各队列，每消费者都接收到了消息。
思考
1、publish/subscribe与work queues有什么区别
区别：
1）work queues不用定义交换机，而publish/subscribe需要定义交换机。
2）publish/subscribe的生产方是面向交换机发送消息，work queues的生产方是面向队列发送消息(底层使用默认
交换机)。
3）publish/subscribe需要设置队列和交换机的绑定，work queues不需要设置，实质上work queues会将队列绑
定到默认的交换机 。
相同点：
两者实现的发布/订阅的效果是一样的，多个消费端监听同一个队列不会重复消费消息
2、工作用什么 publish/subscribe还是work queues。
建议使用 publish/subscribe，发布订阅模式比工作队列模式更强大，并且发布订阅模式可以指定自己专用的交换
机 ，也可以做到work queues模式的效果
生产者全部代码可直接copy
public class Producer03_routing {
//email消息队列
private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
private static final String ROUTINGKEY_EMAIL = "routingkey_email";
//sms消息队列
private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
private static final String ROUTINGKEY_SMS = "routingkey_sms";
//交换机
private static final String EXCHANGE_ROUTING_INFORM = "exchange_routing_inform";
public static void main(String[] args) {
//连接
Connection connection = null;
//通道
Channel channel = null;
try {
//给MQ发送消息
//连接MQ
//通过连接工厂创建连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");//IP地址
connectionFactory.setPort(5672);//默认mq服务端口
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");//设置mq虚拟机
//和MQ创建连接
connection = connectionFactory.newConnection();
//建立channel通道，会话通道，在通道中向mq发送消息
channel = connection.createChannel();
//声明交换机
/**
* String exchange, BuiltinExchangeType type
* 1、交换机名称,采用发布订阅模式，交换机的类型要设置为fanout，Routing路由模式将交换机的类型设置为direct
* 如果使用的是Topics模式将交换机的类型设置为topics。。使用HEADERS模式将交换机类型设置为HEADERS
*/
channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
//声明一个队列,根据队列名称判断，如果在mq中没有此队列就创建一个队列，如果有此队列则不作处理
/**
* 参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
* 1、队列名称
* 2、是否持久化，true表示持久化，当mq重启之后此队列还在
* 3、是否独占通道，true表示此通道只能由此队列来使用
* 4、自动删除，true表示自动删除，mq重启后队列删除
* 5、队列参数列表
*/
channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
//将队列绑定到交换机中,目的就是让交换机将消息转发到队列
/**
* 参数：String queue, String exchange, String routingKey
* 1、队列名称
* 2、交换机名称
* 3、routingKey路由key，交换机根据路由key将消息转发到队列,发布订阅模式不需要指定routingkey
*/
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL);
channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS);
//发送消息
/**
* 参数String exchange, String routingKey, BasicProperties props, byte[] body；
* 1、exchange 交换机名称
* 2、routingKey routing路由模式需要指定routingkey
* 3、消息属性
* 4、消息内容
*/
for(int i=0;i<2;i++){
String message ="send email message:"+ System.currentTimeMillis();
channel.basicPublish(EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL,null,message.getBytes());
System.out.println("send message.."+message);
}
for(int i=0;i<3;i++){
String message ="send sms message:"+ System.currentTimeMillis();
channel.basicPublish(EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS,null,message.getBytes());
System.out.println("send message.."+message);
}
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
} finally {
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
try {
connection.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
消费者全部代码
public class Consumer03_routing_email {
//email消息队列
private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
private static final String ROUTINGKEY_EMAIL = "routingkey_email";
//交换机
private static final String EXCHANGE_ROUTING_INFORM = "exchange_routing_inform";
public static void main(String[] args) {
//连接
Connection connection = null;
//通道
Channel channel = null;
try {
//给MQ发送消息
//连接MQ
//通过连接工厂创建连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");//IP地址
connectionFactory.setPort(5672);//默认mq服务端口
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");//设置mq虚拟机
//和MQ创建连接
connection = connectionFactory.newConnection();
//建立channel通道，会话通道，在通道中向mq发送消息
channel = connection.createChannel();
//声明交换机
/**
* String exchange, BuiltinExchangeType type
* 1、交换机名称,采用发布订阅模式，交换机的类型要设置为fanout，Routing路由模式将交换机的类型设置为direct
* 如果使用的是Topics模式将交换机的类型设置为topics。。使用HEADERS模式将交换机类型设置为HEADERS
*/
channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
//声明一个队列,根据队列名称判断，如果在mq中没有此队列就创建一个队列，如果有此队列则不作处理
/**
* 参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
* 1、队列名称
* 2、是否持久化，true表示持久化，当mq重启之后此队列还在
* 3、是否独占通道，true表示此通道只能由此队列来使用
* 4、自动删除，true表示自动删除，mq重启后队列删除
* 5、队列参数列表
*/
channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
//将队列绑定到交换机中,目的就是让交换机将消息转发到队列
/**
* 参数：String queue, String exchange, String routingKey
* 1、队列名称
* 2、交换机名称
* 3、routingKey路由key，交换机根据路由key将消息转发到队列,发布订阅模式不需要指定routingkey
*/
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL);
//创建回调方法类
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
* 消费者接收到消息后会调用此方法
* @param consumerTag 消费者标签，用来标识消费，如果不指定默认一个名称
* @param envelope 消息内容包
* @param properties 消息的属性
* @param body 消息的内容
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String exchange = envelope.getExchange();//交换
long deliveryTag = envelope.getDeliveryTag();//消息id
String routingKey = envelope.getRoutingKey();//路由key
String message = new String(body, "utf-8");
System.out.println(message);
}
};
//监听队列,接收消息
/**
* 参数：String queue, boolean autoAck, Consumer callback
* 1、监听队列的名称
* 2、autoAck是否自动回复，消息者接收到消息要给mq回复表示此消息已接收，此时mq去删除消息
* 如果autoAck设置true，表示消费者接收到消息后自动回复，如果设置为false，需要程序员在代码中手动回复(channel.basicAck();)
* 3、回调方法，接收到消息后调用此callback
*/
channel.basicConsume(QUEUE_INFORM_EMAIL,true,consumer);
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}
使用生产者发送若干条消息，交换机根据routingkey转发消息到指定的队列
思考
1、Routing模式和Publish/subscibe有啥区别？
Routing模式要求队列在绑定交换机时要指定routingkey，消息会转发到符合routingkey的队列
案例：
根据用户的通知设置去通知用户，设置接收Email的用户只接收Email，设置接收sms的用户只接收sms，设置两种
通知类型都接收的则两种通知都有效。
生产者全部代码
public class Producer04_topics {
//email消息队列
private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
private static final String ROUTINGKEY_EMAIL = "inform.#.email.#";
//sms消息队列
private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
private static final String ROUTINGKEY_SMS = "inform.#.sms.#";
//交换机
private static final String EXCHANGE_TOPICS_INFORM = "exchange_topics_inform";
public static void main(String[] args) {
//连接
Connection connection = null;
//通道
Channel channel = null;
try {
//给MQ发送消息
//连接MQ
//通过连接工厂创建连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");//IP地址
connectionFactory.setPort(5672);//默认mq服务端口
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");//设置mq虚拟机
//和MQ创建连接
connection = connectionFactory.newConnection();
//建立channel通道，会话通道，在通道中向mq发送消息
channel = connection.createChannel();
//声明交换机
/**
* String exchange, BuiltinExchangeType type
* 1、交换机名称,采用发布订阅模式，交换机的类型要设置为fanout，Routing路由模式将交换机的类型设置为direct
* 如果使用的是Topics模式将交换机的类型设置为topics。。使用HEADERS模式将交换机类型设置为HEADERS
*/
channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);
//声明一个队列,根据队列名称判断，如果在mq中没有此队列就创建一个队列，如果有此队列则不作处理
/**
* 参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
* 1、队列名称
* 2、是否持久化，true表示持久化，当mq重启之后此队列还在
* 3、是否独占通道，true表示此通道只能由此队列来使用
* 4、自动删除，true表示自动删除，mq重启后队列删除
* 5、队列参数列表
*/
channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
//将队列绑定到交换机中,目的就是让交换机将消息转发到队列
/**
* 参数：String queue, String exchange, String routingKey
* 1、队列名称
* 2、交换机名称
* 3、routingKey路由key，交换机根据路由key将消息转发到队列,发布订阅模式不需要指定routingkey
*/
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_TOPICS_INFORM,ROUTINGKEY_EMAIL);
channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_TOPICS_INFORM,ROUTINGKEY_SMS);
//发送消息
/**
* 参数String exchange, String routingKey, BasicProperties props, byte[] body；
* 1、exchange 交换机名称
* 2、routingKey routing路由模式需要指定routingkey
* 3、消息属性
* 4、消息内容
*/
//发送email
// for(int i=0;i<2;i++){
// String message ="send email message:"+ System.currentTimeMillis();
// channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.email",null,message.getBytes());
// System.out.println("send message.."+message);
// }
//只发送sms
/*for(int i=0;i<3;i++){
String message ="send sms message:"+ System.currentTimeMillis();
channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.sms",null,message.getBytes());
System.out.println("send message.."+message);
}*/
//发送sms和email
for(int i=0;i<3;i++){
String message ="send sms and email message:"+ System.currentTimeMillis();
channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.sms.email",null,message.getBytes());
System.out.println("send message.."+message);
}
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
} finally {
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
try {
connection.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
邮件消费者全部代码
public class Consumer04_topics_email {
//email消息队列
private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
private static final String ROUTINGKEY_EMAIL = "inform.#.email.#";
//交换机
private static final String EXCHANGE_TOPICS_INFORM = "exchange_topics_inform";
public static void main(String[] args) {
//连接
Connection connection = null;
//通道
Channel channel = null;
try {
//给MQ发送消息
//连接MQ
//通过连接工厂创建连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");//IP地址
connectionFactory.setPort(5672);//默认mq服务端口
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");//设置mq虚拟机
//和MQ创建连接
connection = connectionFactory.newConnection();
//建立channel通道，会话通道，在通道中向mq发送消息
channel = connection.createChannel();
//声明交换机
/**
* String exchange, BuiltinExchangeType type
* 1、交换机名称,采用发布订阅模式，交换机的类型要设置为fanout，Routing路由模式将交换机的类型设置为direct
* 如果使用的是Topics模式将交换机的类型设置为topics。。使用HEADERS模式将交换机类型设置为HEADERS
*/
channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);
//声明一个队列,根据队列名称判断，如果在mq中没有此队列就创建一个队列，如果有此队列则不作处理
/**
* 参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
* 1、队列名称
* 2、是否持久化，true表示持久化，当mq重启之后此队列还在
* 3、是否独占通道，true表示此通道只能由此队列来使用
* 4、自动删除，true表示自动删除，mq重启后队列删除
* 5、队列参数列表
*/
channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
//将队列绑定到交换机中,目的就是让交换机将消息转发到队列
/**
* 参数：String queue, String exchange, String routingKey
* 1、队列名称
* 2、交换机名称
* 3、routingKey路由key，交换机根据路由key将消息转发到队列,发布订阅模式不需要指定routingkey
*/
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_TOPICS_INFORM,ROUTINGKEY_EMAIL);
//创建回调方法类
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
* 消费者接收到消息后会调用此方法
* @param consumerTag 消费者标签，用来标识消费，如果不指定默认一个名称
* @param envelope 消息内容包
* @param properties 消息的属性
* @param body 消息的内容
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String exchange = envelope.getExchange();//交换
long deliveryTag = envelope.getDeliveryTag();//消息id
String routingKey = envelope.getRoutingKey();//路由key
String message = new String(body, "utf-8");
System.out.println(message);
}
};
//监听队列,接收消息
/**
* 参数：String queue, boolean autoAck, Consumer callback
* 1、监听队列的名称
* 2、autoAck是否自动回复，消息者接收到消息要给mq回复表示此消息已接收，此时mq去删除消息
* 如果autoAck设置true，表示消费者接收到消息后自动回复，如果设置为false，需要程序员在代码中手动回复(channel.basicAck();)
* 3、回调方法，接收到消息后调用此callback
*/
channel.basicConsume(QUEUE_INFORM_EMAIL,true,consumer);
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}
测试
使用生产者发送若干条消息，交换机根据routingkey统配符匹配并转发消息到指定的队列
队列绑定交换机指定通配符：
统配符规则：
中间以“.”分隔。
符号#可以匹配多个词，符号*可以匹配一个词语。