【今日感悟】
本文只介绍通配符类型，想了解更多类型自行浏览官网，
RabbitMQ工作模式-订阅模式类型介绍
- P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
- C：消费者，消息的接受者，会一直等待消息到来。
- Queue：消息队列，接收消息、缓存消息。
- Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如把消息递交给某个特别队列、递交给所有队列、或是将消息丢弃。 到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：
- Fanout：广播，将消息交给所有绑定到交换机的队列
- Direct：定向，把消息交给符合指定routing key 的队列
-
Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
1564888319550.png
1563518250036.png
windows环境：
- 安装erlang；
- 安装RabbitMQ；
- 安装管理插件
打开cmd在rabbitMq的sbin中输入下面命令
rabbitmq-plugins.bat enable rabbitmq_management - 创建管理RabbitMQ的用户http://localhost:15672
- 创建虚拟主机Virtual Hosts
连接工具类：
public class ConnectionUtil {
public static Connection getConnection() throws Exception {
//创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//主机地址;默认为 localhost
connectionFactory.setHost("localhost");
//连接端口;默认为 5672
connectionFactory.setPort(5672);
//虚拟主机名称;默认为 /
connectionFactory.setVirtualHost("/");
//连接用户名；默认为guest
connectionFactory.setUsername("guest");
//连接密码；默认为guest
connectionFactory.setPassword("guest");
//创建连接
Connection connection = connectionFactory.newConnection();
return connection;
}
}
消息生产者：
/**
* Topic通配符模型消息生产者：消息发送到交换机
*/
public class Producer {
// 交换机名称
public static final String TOPIC_EXCHANGE = "topic_exchange";
// 队列名称
public static final String TOPIC_QUEUE_1 = "topic_queue_1";
// 队列名称
public static final String TOPIC_QUEUE_2 = "topic_queue_2";
public static void main(String[] args) throws Exception {
// 1.创建连接
Connection connection = ConnectionUtil.getConnection();
// 2.创建频道
Channel channel = connection.createChannel();
/**
* 3.声明交换机
* 参数1：交换机名称
* 参数2：交换机类型：fanout、direct、topic、headers
*/
channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC);
// 6.发送消息
String message = "新增了商品，Topic模式，路由key为item.insert";
/**
* 参数1：交换机名称，如果没有指定则使用默认Default Exchage
* 参数2：路由key,简单模式可以传递队列名称
* 参数3：消息其它属性
* 参数4：消息内容
*/
channel.basicPublish(TOPIC_EXCHANGE,"item.insert",null,message.getBytes());
System.out.println("已发送消息：" + message);
message = "修改了商品，Topic模式，路由key为item.update";
channel.basicPublish(TOPIC_EXCHANGE,"item.update",null,message.getBytes());
System.out.println("已发送消息：" + message);
message = "删除了商品，Topic模式，路由key为item.delete";
channel.basicPublish(TOPIC_EXCHANGE,"item.delete",null,message.getBytes());
System.out.println("已发送消息：" + message);
// 7.关闭资源
channel.close();
connection.close();
}
}
消息消费者：
/**
* 消息消费者: 消费队列中的消息
*/
public class Consumer1 {
public static void main(String[] args) throws Exception {
// 1.创建连接
Connection connection = ConnectionUtil.getConnection();
// 2.创建频道
Channel channel = connection.createChannel();
// 3.创建交换机
channel.exchangeDeclare(Producer.TOPIC_EXCHANGE,BuiltinExchangeType.TOPIC);
/**
* 4. 声明（创建）队列
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null);
// 5.队列绑定到交换机
channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHANGE,"item.*");
// 6.创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
/**
* consumerTag 消费者标签，在channel.basicConsume时候可以指定
* envelope 消息包的内容，可从中获取消息id，
* 消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
* properties 属性信息
* body 消息
*/
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("路由key为：" + envelope.getRoutingKey());
System.out.println("交换机为：" + envelope.getExchange());
System.out.println("消息id为：" + envelope.getDeliveryTag());
System.out.println("接收到的消息为：" + new String(body, "utf-8"));
}
};
//监听消息
/**
* 参数1：队列名称
* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，
* mq接收到回复会删除消息，设置为false则需要手动确认
* 参数3：消息接收到后回调
*/
channel.basicConsume(Producer.TOPIC_QUEUE_1, true, consumer);
//不关闭资源，应该一直监听消息
}
}
