3 工作模式
3.1 Work queues
3.1.1 特点
- work queues与入门程序相比，多了一个消费端，两个消费端共同消费同一个队列中的消息。
入门程序启动多个consumer即可实现。
- 一条消息只会被一个消费者接收；
- rabbit采用轮询（你一个我一个）的方式将消息是平均发送给消费者的；
- 消费者在处理完某条消息后，才会收到下一条消息。
3.2 Publish/Subscribe 发布订阅（不需要指定routingKey）
3.2.1 特点
- 每个消费者监听自己的队列。
- 生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，<font color="red">每个绑定交换机的队列都将接收到消息</font>
3.2.2 代码
- 案例场景：用户通知，当用户充值成功或转账完成系统通知用户，通知方式有短信、邮件多种方法 。
3.2.2.1 生产者
- 声明Exchange_fanout_inform交换机。
- 声明两个队列并且绑定到此交换机，绑定时不需要指定routingkey
- <font color="red">发送消息时不需要指定routingkey</font>
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;
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);//外部web管理端口时 15672，注意区分
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
//设置虚拟机,一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
connectionFactory.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 是否独占连接，队列只允许在该连接中访问，
* 如果连接关闭，队列自动删除；如果将此参数设置为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.exchange 交换机名称
* 2.type 交换机的类型
* fanout:对应的rabbitmq的工作模式是 public / 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 < 5; i++) {
String message = "发布通知消息" + i;
channel.basicPublish(EXCHANGE_FANOUT_INFORM, "", null, message.getBytes());
System.out.println("发送给mq的信息为：" + 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();
}
}
}
}3.2.2.2 消费者
- 声明交换机要指定交换机类型为
fanout，BuiltinExchangeType.FANOUT
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
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);//外部web管理端口时 15672，注意区分
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
//设置虚拟机,一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
connectionFactory.setVirtualHost("/");
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 是否独占连接，队列只允许在该连接中访问，
* 如果连接关闭，队列自动删除；如果将此参数设置为true可用于临时队列的创建
* 4.autoDelete 自动删除，队列不再是哟个时是否自动删除此队列
* 如果将此参数和exclusive参数设置为true,就可以实现临时队列（队列不用了就自动删除）
* 5.arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间等
*/
channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
//声明交换机
//参数：String exchange, String type
/**
* 1.exchange 交换机名称
* 2.type 交换机的类型
* fanout:对应的rabbitmq的工作模式是 public / 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 可得到交换机和消息id等
* @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();
//消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
long deliveryTag = envelope.getDeliveryTag();
//消息内容
String message = new String(body, "utf-8");
System.out.println("接受到的消息：" + message);
}
};
//监听队列
//参数：String queue, boolean autoAck, Consumer callback
/**
* 1.queue 队列名称
* 2.autoAck 自动回复
* 当消费者接收到消息后要告诉mq消息已经接收
* true表示自动回复mq
* false表示要通过编程实现回复
* 3.callback 消费方法，当消费者接收到消息要执行的方法
*/
channel.basicConsume(QUEUE_INFORM_EMAIL, true, defaultConsumer);
}
}3.2.3 publish/subscribe与work queues的区异同
区别
- work queues不用定义交换机，而publish/subscribe需要定义交换机。
publish/subscribe的生产方是面向交换机发送消息，work queues的生产方是面向队列发送消息(底层使用默认交换机)。publish/subscribe需要设置队列和交换机的绑定，work queues不需要设置，实质上work queues会将队列绑定到默认的交换机 。
实质工作用什么
publish/subscribe还是work queues- 建议使用
publish/subscribe，发布订阅模式比工作队列模式更强大，并且发布订阅模式可以指定自己专用的交换机
- 建议使用
3.3 Routing 路由 (设置routingKey)
3.3.1 特点
- 每个消费者监听自己的队列，并且设置
routingkey。 - 生产者将消息发给交换机，由交换机根据
routingkey来转发消息到指定的队列。
3.3.2 代码
3.3.2.1 生产者
- 声明exchange_routing_inform交换机。
- 声明两个队列并且绑定到此交换机，<font color="red">绑定时需要指定
routingKey</font> - <font color="red">发送消息时需要指定
routingkey</font>
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;
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";
//routingKey 邮件key的消息传给邮件的queue，短信key的消息传给短信的queue
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);//外部web管理端口时 15672，注意区分
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
//设置虚拟机,一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
connectionFactory.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 是否独占连接，队列只允许在该连接中访问，
* 如果连接关闭，队列自动删除；如果将此参数设置为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.exchange 交换机名称
* 2.type 交换机的类型
* fanout:对应的rabbitmq的工作模式是 public / 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");//1个队列绑定多个routingKey
channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_ROUTING_INFORM, ROUTINGKEY_SMS);
channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_ROUTING_INFORM, "inform");//不同队列绑定相同routingKey，两个队列均可接收同一routingKey的消息
//发送消息
//参数：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++) {
//发送消息的时候指定routingKey
String message = "发布email通知消息" + i;
channel.basicPublish(EXCHANGE_ROUTING_INFORM, ROUTINGKEY_EMAIL, null, message.getBytes());
System.out.println("发送给email的信息为：" + message);
}
for (int i = 0; i < 5; i++) {
//发送消息的时候指定routingKey
String message = "发布sms通知消息" + i;
channel.basicPublish(EXCHANGE_ROUTING_INFORM, ROUTINGKEY_SMS, null, message.getBytes());
//给相同routingKey的队列发消息
// channel.basicPublish(EXCHANGE_ROUTING_INFORM, "inform", null, message.getBytes());
System.out.println("发送给sms的信息为：" + 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();
}
}
}
}3.3.2.2 消费者
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
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";
//routingKey 邮件key的消息传给邮件的queue，短信key的消息传给短信的queue
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);//外部web管理端口时 15672，注意区分
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
//设置虚拟机,一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
connectionFactory.setVirtualHost("/");
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 是否独占连接，队列只允许在该连接中访问，
* 如果连接关闭，队列自动删除；如果将此参数设置为true可用于临时队列的创建
* 4.autoDelete 自动删除，队列不再是哟个时是否自动删除此队列
* 如果将此参数和exclusive参数设置为true,就可以实现临时队列（队列不用了就自动删除）
* 5.arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间等
*/
channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
//声明交换机
//参数：String exchange, String type
/**
* 1.exchange 交换机名称
* 2.type 交换机的类型
* fanout:对应的rabbitmq的工作模式是 public / 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 可得到交换机和消息id等
* @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();
//消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
long deliveryTag = envelope.getDeliveryTag();
//消息内容
String message = new String(body, "utf-8");
System.out.println("接受到的消息：" + message);
}
};
//监听队列
//参数：String queue, boolean autoAck, Consumer callback
/**
* 1.queue 队列名称
* 2.autoAck 自动回复
* 当消费者接收到消息后要告诉mq消息已经接收
* true表示自动回复mq
* false表示要通过编程实现回复
* 3.callback 消费方法，当消费者接收到消息要执行的方法
*/
channel.basicConsume(QUEUE_INFORM_EMAIL,true, defaultConsumer);
}
}3.3.3 Routing模式和Publish/subscibe的区别
- Routing模式要求队列在绑定交换机时要指定
routingkey，消息会转发到符合routingkey的队列。
3.4 Topics 通配符（消费者通配设置routingKey）
3.4.1 特点
- 每个消费者监听自己的队列，并且设置带统配符的
routingkey。 - 生产者将消息发给broker，由交换机根据
routingkey来转发消息到指定的队列。
3.4.2 代码
- 案例：根据用户的通知设置去通知用户，设置接收Email的用户只接收Email，设置接收sms的用户只接收sms，设置两种通知类型都接收的则两种通知都有效。
3.4.2.1 生产者
- <font color="red">声明交换机，指定topic类型</font>
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;
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";
//routingKey 邮件key的消息传给邮件的queue，短信key的消息传给短信的queue
private static final String ROUTINGKEY_EMAIL = "inform.#.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);//外部web管理端口时 15672，注意区分
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
//设置虚拟机,一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
connectionFactory.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 是否独占连接，队列只允许在该连接中访问，
* 如果连接关闭，队列自动删除；如果将此参数设置为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.exchange 交换机名称
* 2.type 交换机的类型
* fanout:对应的rabbitmq的工作模式是 public / 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++) {
//发送消息的时候指定routingKey
String message = "发布email通知消息" + i;
channel.basicPublish(EXCHANGE_TOPICS_INFORM, "inform.email", null, message.getBytes());
System.out.println("发送给email的信息为：" + message);
}
for (int i = 0; i < 5; i++) {
//发送消息的时候指定routingKey
String message = "发布sms通知消息" + i;
channel.basicPublish(EXCHANGE_TOPICS_INFORM, "inform.sms", null, message.getBytes());
System.out.println("发送给sms的信息为：" + message);
}
for (int i = 0; i < 5; i++) {
//发送消息的时候指定routingKey
String message = "发布email和sms通知消息" + i;
channel.basicPublish(EXCHANGE_TOPICS_INFORM, "inform.email.sms", null, message.getBytes());
System.out.println("发送给email和sms的信息为：" + 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();
}
}
}
}3.4.2.2 消费者
队列绑定交换机指定通配符：
统配符规则：
中间以“.”分隔。
符号#可以匹配多个词 inform.email.sms inform.email均可
符号*可以匹配一个词 infor.email。- 绑定队列进行修改
private static final String ROUTINGKEY_EMAIL = "inform.#.email.#";//inform.email
//省略部分。。。
//交换机和队列进行绑定
//参数：String queue, String exchange, String routingKey
/**
* 1.queue 队列名称
* 2.exchange 交换机名称
* 3.routingKey 路由key，在发布订阅模式中设置为空字符串
* 作用：交换机根据路由Key的值将消息转发到指定的队列中
*/
channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_TOPICS_INFORM, ROUTINGKEY_EMAIL);3.5 Header模式
- header模式与routing不同的地方在于，header模式取消
routingkey，使用header中的 key/value（键值对）匹配队列。
3.5.1 代码
3.5.1.1 生产者
//新建map存放匹配队列的key/value
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");
//绑定队列和交换机时，不需要指定routingKey，
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_HEADERS_INFORM,"",headers_email);
channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_HEADERS_INFORM,"",headers_sms);
//...
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());3.5.1.2 消费者
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);3.6 RPC
RPC即客户端远程调用服务端的方法 ，使用MQ可以实现RPC的异步调用，基于Direct交换机实现，流程如下：
- 客户端即是生产者就是消费者，向RPC请求队列发送RPC调用消息，同时监听RPC响应队列。
- 服务端监听RPC请求队列的消息，收到消息后执行服务端的方法，得到方法返回的结果
- 服务端将RPC方法 的结果发送到RPC响应队列
- 客户端（RPC调用方）监听RPC响应队列，接收到RPC调用结果。
4 SpringBoot整合RabbitMQ
4.1 环境搭建
- Maven中添加依赖
spring-boot-starter-amqp
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
</dependency>4.2 配置
- 生产者项目和消费者项目都需要
4.2.1 application.yml
server:
port: 44000
spring:
application:
name: test-rabbit-producer
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
virtual-host: /4.2.2 RabbitConfig类
- 配置
exchange交换机、Queue队列、绑定交换机
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.#";//inform.email
public static final String ROUTINGKEY_SMS = "inform.#.sms.#";
//声明交换机
@Bean(EXCHANGE_TOPICS_INFORM)
public Exchange getTopicsExchange() {
//durable(true) 持久化，mq重启后交换机还在
return ExchangeBuilder.topicExchange(EXCHANGE_TOPICS_INFORM).durable(true).build();
}
//声明队列 QUEUE_INFORM_EMAIL
@Bean(QUEUE_INFORM_EMAIL)
public Queue getQueueInformEmail() {
return new Queue(QUEUE_INFORM_EMAIL);
}
//声明队列 QUEUE_INFORM_SMS
@Bean(QUEUE_INFORM_SMS)
public Queue getQueueInformSms() {
return new Queue(QUEUE_INFORM_SMS);
}
//绑定交换机和email队列,指定routingKey
@Bean
public Binding bindingQueueInformEmail(@Qualifier(QUEUE_INFORM_EMAIL) Queue queue,
@Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_EMAIL).noargs();
}
//绑定交换机和sms队列,指定routingKey
@Bean
public Binding bindingQueueInformSms(@Qualifier(QUEUE_INFORM_SMS) Queue queue,
@Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_SMS).noargs();
}
}
4.3 生产者
- 使用
RabbitTemplate发送消息
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.xuecheng.test.rabbitmq.config.RabbitmqConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
@SpringBootTest
@RunWith(SpringRunner.class)
public class Producer04_topics_springboot {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSendEmail() {
String message = "给email发送消息";
//参数
/**
* 1.交换机的名称
* 2.routingKey
* 3.消息内容
*/
rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_TOPICS_INFORM, "inform.email", message);
System.out.println(message);
}
}
4.4 消费端
- 使用
@RabbitListener注解监听队列
import com.rabbitmq.client.Channel;
import com.xuecheng.test.rabbitmq.config.RabbitmqConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @author hh
* @create 2020/2/4
* @description
**/
@Component
public class ReceiveHandler {
@RabbitListener(queues = {RabbitmqConfig.QUEUE_INFORM_EMAIL})
public void send_email(String message, Message message2, Channel channel) {
System.out.println("接收到的消息" + message);
System.out.println("接收到的消息" + message2.getBody());
}
}