文章目录

2_RabbitMQ工作模式
- 1.Work queues
- 2.Publish/Subscribe
- - 1.工作模式
  - 2.代码
  - - 1.生产者
    - - 1.指定消息队列相关消息
      - 2.建立连接&绑定队列
      - 3.发送消息
      - 完整代码：
    - 2.消费者
    - - 1.指定消息队列相关消息
      - 2.建立连接&绑定队列
      - 3.实现消费方法&监听消息
      - 完整代码
  - 3.小结
- 3.Routing
- - 1.工作模式
  - 2.代码
  - - 1.生产者
    - 2.消费者
  - 3.小结
- 4.Topics
- - 1.工作模式
  - 2.代码
  - - 1.生产者
    - 2.消费者
  - 3.小结
- 5.Header&Rpc
- - header模式
  - - 1）生产者
    - 2）发送邮件消费者
  - Rpc

2_RabbitMQ工作模式

前置知识：工作模式指的是某条消息该怎么分发到队列？分发策略

1.Work queues

work queues与入门程序相比，多了一个消费端，两个消费端共同消费同一个队列中的消息。

应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

测试：

1、使用入门程序，启动多个消费者。

2、生产者发送多个消息。

结果：

1、一条消息只会被一个消费者接收；

2、rabbit采用轮询的方式将消息是平均发送给消费者的；

3、消费者在处理完某条消息后，才会收到下一条消息。

2.Publish/Subscribe

1.工作模式

发布订阅模式：

1、每个消费者监听自己的队列。

2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息

此交换机的概念类似计算机网络中的交换机，接受包然后分发下去

2.代码

案例：

用户通知，当用户充值成功或转账完成系统通知用户，通知方式有短信、邮件多种方法。

1.生产者

声明exchange_routing_inform交换机。

声明两个队列并且绑定到此交换机，绑定时需要指定routingkey

发送消息时需要指定routingkey

1.指定消息队列相关消息

//队列名称
private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
private static final String EXCHANGE_FANOUT_INFORM="exchange_fanout_inform";//通过连接工厂创建新的连接和mq建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mqconnectionFactory.setVirtualHost("/");

2.建立连接&绑定队列

Connection connection = null;
Channel channel = null;//建立新连接
connection = connectionFactory.newConnection();
//创建会话通道,生产者和mq服务所有通信都在channel通道中完成
channel = connection.createChannel();
//声明队列，如果队列在mq 中没有则要创建
//参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化，如果持久化，mq重启后队列还在* 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建* 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）* 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间*/
channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
//声明一个交换机
//参数：String exchange, String type
/*** 参数明细：* 1、交换机的名称* 2、交换机的类型* fanout：对应的rabbitmq的工作模式是 publish/subscribe* direct：对应的Routing	工作模式* topic：对应的Topics工作模式* headers： 对应的headers工作模式*/
channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);
//进行交换机和队列绑定
//参数：String queue, String exchange, String routingKey
/*** 参数明细：* 1、queue 队列名称* 2、exchange 交换机名称* 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串*/
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,"");
channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_FANOUT_INFORM,"");

3.发送消息

//发送消息
//参数：String exchange, String routingKey, BasicProperties props, byte[] body
/*** 参数明细：* 1、exchange，交换机，如果不指定将使用mq的默认交换机（设置为""）* 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称* 3、props，消息的属性* 4、body，消息内容*/
for(int i=0;i<20;i++){//消息内容String message = "send inform message to user "+(i+1);channel.basicPublish(EXCHANGE_FANOUT_INFORM,"",null,message.getBytes());System.out.println("send to mq "+message);
}

完整代码：

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Administrator* @version 1.0* @create 2018-06-17 18:10**/
public class Producer02_publish {//队列名称private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";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) {//通过连接工厂创建新的连接和mq建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mqconnectionFactory.setVirtualHost("/");Connection connection = null;Channel channel = null;try {//建立新连接connection = connectionFactory.newConnection();//创建会话通道,生产者和mq服务所有通信都在channel通道中完成channel = connection.createChannel();//声明队列，如果队列在mq 中没有则要创建//参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化，如果持久化，mq重启后队列还在* 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建* 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）* 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);//声明一个交换机//参数：String exchange, String type/*** 参数明细：* 1、交换机的名称* 2、交换机的类型* fanout：对应的rabbitmq的工作模式是 publish/subscribe* direct：对应的Routing	工作模式* topic：对应的Topics工作模式* headers： 对应的headers工作模式*/channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);//进行交换机和队列绑定//参数：String queue, String exchange, String routingKey/*** 参数明细：* 1、queue 队列名称* 2、exchange 交换机名称* 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串*/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，交换机，如果不指定将使用mq的默认交换机（设置为""）* 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称* 3、props，消息的属性* 4、body，消息内容*/for(int i=0;i<20;i++){//消息内容String message = "send inform message to user "+(i+1);channel.basicPublish(EXCHANGE_FANOUT_INFORM,"",null,message.getBytes());System.out.println("send to mq "+message);}} catch (Exception 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();}}}
}

2.消费者

1.指定消息队列相关消息

//队列名称
private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
private static final String EXCHANGE_FANOUT_INFORM="exchange_fanout_inform";//通过连接工厂创建新的连接和mq建立连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);//端口
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
//设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
connectionFactory.setVirtualHost("/");

2.建立连接&绑定队列

Connection connection = null;
Channel channel = null;//建立新连接
connection = connectionFactory.newConnection();
//创建会话通道,生产者和mq服务所有通信都在channel通道中完成
channel = connection.createChannel();
//声明队列，如果队列在mq 中没有则要创建
//参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化，如果持久化，mq重启后队列还在* 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建* 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）* 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间*/
channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
//声明一个交换机
//参数：String exchange, String type
/*** 参数明细：* 1、交换机的名称* 2、交换机的类型* fanout：对应的rabbitmq的工作模式是 publish/subscribe* direct：对应的Routing	工作模式* topic：对应的Topics工作模式* headers： 对应的headers工作模式*/
channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);
//进行交换机和队列绑定
//参数：String queue, String exchange, String routingKey
/*** 参数明细：* 1、queue 队列名称* 2、exchange 交换机名称* 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串*/
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,"");
channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_FANOUT_INFORM,"");

3.实现消费方法&监听消息

//实现消费方法
DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 当接收到消息后此方法将被调用* @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume* @param envelope 信封，通过envelope* @param properties 消息属性* @param body 消息内容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交换机String exchange = envelope.getExchange();//消息id，mq在channel中用来标识消息的id，可用于确认消息已接收long deliveryTag = envelope.getDeliveryTag();//消息内容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}
};
//监听队列
//参数：String queue, boolean autoAck, Consumer callback
/*** 参数明细：* 1、queue 队列名称* 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复* 3、callback，消费方法，当消费者接收到消息要执行的方法*/
channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);

完整代码

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Administrator* @version 1.0* @create 2018-06-17 18:22**/
public class Consumer02_subscribe_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) throws IOException, TimeoutException {//通过连接工厂创建新的连接和mq建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mqconnectionFactory.setVirtualHost("/");//建立新连接Connection connection = connectionFactory.newConnection();//创建会话通道,生产者和mq服务所有通信都在channel通道中完成Channel channel = connection.createChannel();/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化，如果持久化，mq重启后队列还在* 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建* 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）* 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);//声明一个交换机//参数：String exchange, String type/*** 参数明细：* 1、交换机的名称* 2、交换机的类型* fanout：对应的rabbitmq的工作模式是 publish/subscribe* direct：对应的Routing   工作模式* topic：对应的Topics工作模式* headers： 对应的headers工作模式*/channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);//进行交换机和队列绑定//参数：String queue, String exchange, String routingKey/*** 参数明细：* 1、queue 队列名称* 2、exchange 交换机名称* 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, "");//实现消费方法DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 当接收到消息后此方法将被调用* @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume* @param envelope 信封，通过envelope* @param properties 消息属性* @param body 消息内容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交换机String exchange = envelope.getExchange();//消息id，mq在channel中用来标识消息的id，可用于确认消息已接收long deliveryTag = envelope.getDeliveryTag();//消息内容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}};//监听队列//参数：String queue, boolean autoAck, Consumer callback/*** 参数明细：* 1、queue 队列名称* 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复* 3、callback，消费方法，当消费者接收到消息要执行的方法*/channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);}
}

3.小结

3.Routing

1.工作模式

路由模式：

1、每个消费者监听自己的队列，并且设置routingkey。

2、生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列。

2.代码

1.生产者

全部代码：

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Administrator* @version 1.0* @create 2018-06-17 19:23**/
public class Producer03_routing {//队列名称private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";private static final String QUEUE_INFORM_SMS = "queue_inform_sms";private static final String EXCHANGE_ROUTING_INFORM="exchange_routing_inform";private static final String ROUTINGKEY_EMAIL="inform_email";private static final String ROUTINGKEY_SMS="inform_sms";public static void main(String[] args) {//通过连接工厂创建新的连接和mq建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mqconnectionFactory.setVirtualHost("/");Connection connection = null;Channel channel = null;try {//建立新连接connection = connectionFactory.newConnection();//创建会话通道,生产者和mq服务所有通信都在channel通道中完成channel = connection.createChannel();//声明队列，如果队列在mq 中没有则要创建//参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化，如果持久化，mq重启后队列还在* 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建* 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）* 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);//声明一个交换机//参数：String exchange, String type/*** 参数明细：* 1、交换机的名称* 2、交换机的类型* fanout：对应的rabbitmq的工作模式是 publish/subscribe* direct：对应的Routing   工作模式* topic：对应的Topics工作模式* headers： 对应的headers工作模式*/channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);//进行交换机和队列绑定//参数：String queue, String exchange, String routingKey/*** 参数明细：* 1、queue 队列名称* 2、exchange 交换机名称* 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL);channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM,"inform");channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS);channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM,"inform");//发送消息//参数：String exchange, String routingKey, BasicProperties props, byte[] body/*** 参数明细：* 1、exchange，交换机，如果不指定将使用mq的默认交换机（设置为""）* 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称* 3、props，消息的属性* 4、body，消息内容*//* for(int i=0;i<5;i++){//发送消息的时候指定routingKeyString message = "send email inform message to user";channel.basicPublish(EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL,null,message.getBytes());System.out.println("send to mq "+message);}for(int i=0;i<5;i++){//发送消息的时候指定routingKeyString message = "send sms inform message to user";channel.basicPublish(EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS,null,message.getBytes());System.out.println("send to mq "+message);}*/for(int i=0;i<5;i++){//发送消息的时候指定routingKeyString message = "send inform message to user";channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform",null,message.getBytes());System.out.println("send to mq "+message);}} catch (Exception 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();}}}
}

2.消费者

全部代码：

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Administrator* @version 1.0* @create 2018-06-17 18:22**/
public class Consumer03_routing_email {//队列名称private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";private static final String EXCHANGE_ROUTING_INFORM="exchange_routing_inform";private static final String ROUTINGKEY_EMAIL="inform_email";public static void main(String[] args) throws IOException, TimeoutException {//通过连接工厂创建新的连接和mq建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mqconnectionFactory.setVirtualHost("/");//建立新连接Connection connection = connectionFactory.newConnection();//创建会话通道,生产者和mq服务所有通信都在channel通道中完成Channel channel = connection.createChannel();/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化，如果持久化，mq重启后队列还在* 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建* 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）* 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);//声明一个交换机//参数：String exchange, String type/*** 参数明细：* 1、交换机的名称* 2、交换机的类型* fanout：对应的rabbitmq的工作模式是 publish/subscribe* direct：对应的Routing   工作模式* topic：对应的Topics工作模式* headers： 对应的headers工作模式*/channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);//进行交换机和队列绑定//参数：String queue, String exchange, String routingKey/*** 参数明细：* 1、queue 队列名称* 2、exchange 交换机名称* 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL);//实现消费方法DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 当接收到消息后此方法将被调用* @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume* @param envelope 信封，通过envelope* @param properties 消息属性* @param body 消息内容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交换机String exchange = envelope.getExchange();//消息id，mq在channel中用来标识消息的id，可用于确认消息已接收long deliveryTag = envelope.getDeliveryTag();//消息内容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}};//监听队列//参数：String queue, boolean autoAck, Consumer callback/*** 参数明细：* 1、queue 队列名称* 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复* 3、callback，消费方法，当消费者接收到消息要执行的方法*/channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);}
}

3.小结

4.Topics

1.工作模式

2.代码

1.生产者

声明exchange_routing_inform交换机。

声明两个队列并且绑定到此交换机，绑定时需要指定routingkey

发送消息时需要指定routingkey

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Administrator* @version 1.0* @create 2018-06-17 19:23**/
public class Producer04_topics {//队列名称private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";private static final String QUEUE_INFORM_SMS = "queue_inform_sms";private static final String EXCHANGE_TOPICS_INFORM="exchange_topics_inform";private static final String ROUTINGKEY_EMAIL="inform.#.email.#";private static final String ROUTINGKEY_SMS="inform.#.sms.#";public static void main(String[] args) {//通过连接工厂创建新的连接和mq建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mqconnectionFactory.setVirtualHost("/");Connection connection = null;Channel channel = null;try {//建立新连接connection = connectionFactory.newConnection();//创建会话通道,生产者和mq服务所有通信都在channel通道中完成channel = connection.createChannel();//声明队列，如果队列在mq 中没有则要创建//参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化，如果持久化，mq重启后队列还在* 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建* 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）* 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);//声明一个交换机//参数：String exchange, String type/*** 参数明细：* 1、交换机的名称* 2、交换机的类型* fanout：对应的rabbitmq的工作模式是 publish/subscribe* direct：对应的Routing   工作模式* topic：对应的Topics工作模式* headers： 对应的headers工作模式*/channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);//进行交换机和队列绑定//参数：String queue, String exchange, String routingKey/*** 参数明细：* 1、queue 队列名称* 2、exchange 交换机名称* 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串*/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，交换机，如果不指定将使用mq的默认交换机（设置为""）* 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称* 3、props，消息的属性* 4、body，消息内容*/for(int i=0;i<5;i++){//发送消息的时候指定routingKeyString message = "send email inform message to user";channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.email",null,message.getBytes());System.out.println("send to mq "+message);}for(int i=0;i<5;i++){//发送消息的时候指定routingKeyString message = "send sms inform message to user";channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.sms",null,message.getBytes());System.out.println("send to mq "+message);}for(int i=0;i<5;i++){//发送消息的时候指定routingKeyString message = "send sms and email inform message to user";channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.sms.email",null,message.getBytes());System.out.println("send to mq "+message);}} catch (Exception 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();}}}
}

2.消费者

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** @author Administrator* @version 1.0* @create 2018-06-17 18:22**/
public class Consumer04_topics_email {//队列名称private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";private static final String EXCHANGE_TOPICS_INFORM="exchange_topics_inform";private static final String ROUTINGKEY_EMAIL="inform.#.email.#";public static void main(String[] args) throws IOException, TimeoutException {//通过连接工厂创建新的连接和mq建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mqconnectionFactory.setVirtualHost("/");//建立新连接Connection connection = connectionFactory.newConnection();//创建会话通道,生产者和mq服务所有通信都在channel通道中完成Channel channel = connection.createChannel();/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化，如果持久化，mq重启后队列还在* 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建* 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）* 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);//声明一个交换机//参数：String exchange, String type/*** 参数明细：* 1、交换机的名称* 2、交换机的类型* fanout：对应的rabbitmq的工作模式是 publish/subscribe* direct：对应的Routing   工作模式* topic：对应的Topics工作模式* headers： 对应的headers工作模式*/channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);//进行交换机和队列绑定//参数：String queue, String exchange, String routingKey/*** 参数明细：* 1、queue 队列名称* 2、exchange 交换机名称* 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_TOPICS_INFORM,ROUTINGKEY_EMAIL);//实现消费方法DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 当接收到消息后此方法将被调用* @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume* @param envelope 信封，通过envelope* @param properties 消息属性* @param body 消息内容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交换机String exchange = envelope.getExchange();//消息id，mq在channel中用来标识消息的id，可用于确认消息已接收long deliveryTag = envelope.getDeliveryTag();//消息内容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}};//监听队列//参数：String queue, boolean autoAck, Consumer callback/*** 参数明细：* 1、queue 队列名称* 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复* 3、callback，消费方法，当消费者接收到消息要执行的方法*/channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);}
}