Exchange：接收消息，并根据路由键转发消息所绑定的队列
ClientA,B将消息投递到交换机Exchange上，通过路由关系，投递到指定的queue1或者queue2上，通过监听投递到Client1...
交换机属性:
Name：交换机名称
Type：交换机类型direct、topic、fanout、headers
Durability：是否需要持久化，true为持久化
Auto Delete：当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange
Internal：当前Exchange是否用于RabbitMQ内部使用，默认为false
Arguments：扩展参数，用于扩展AMQP协议自制定化使用
有直连对应关系的连接方式：
Direct Exchange：所有发送到Direct Exchange的消息被转发到RoutKey中指定的Queue
注：Direct模式可以使用RabbitMQ自带的Exchange：default Exchange，
所以不需要将Exchange进行任何绑定（binding）操作，消息加投递时，
RouteKey必须完全匹配才会被队列接收，否则该消息会被抛弃。
routing key要和 queues中的key的值相同,才可以路由到
//生产端代码
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
String exchangeName = "test_direct_exchange";
String routingKey = "test.direct";
String str = "hello world";
channel.basicPublish(exchangeName, routingKey, null, str.getBytes()); //消费端代码
ConnectionFactory connectionFactory = new ConnectionFactory() ;
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
connectionFactory.setAutomaticRecoveryEnabled(true);
connectionFactory.setNetworkRecoveryInterval(3000);
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
//4 声明
String exchangeName = "test_direct_exchange";
String exchangeType = "direct";
String queueName = "test_direct_queue";
String routingKey = "test.direct";
//表示声明了一个交换机
channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
//表示声明了一个队列
channel.queueDeclare(queueName, false, false, false, null);
//建立一个绑定关系:
channel.queueBind(queueName, exchangeName, routingKey);
//durable 是否持久化消息
QueueingConsumer consumer = new QueueingConsumer(channel);
//参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(queueName, true, consumer);
//循环获取消息
while(true){
//获取消息，如果没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery();
String msg = new String(delivery.getBody());
System.out.println("收到消息：" + msg);
} 有路由规则，可模糊匹配的连接方式：
Topic Exchange：所有发送到Topic Exchange的消息被转发到所有关心RouteKey中指定的Topic的Queue上
Exchange将RouteKey和某Topic进行模糊匹配，此时队列需要绑定一个Topic
注：可以使用通配符进行模糊匹配
符号“#”匹配一个或多个词
符号"*"匹配一个词
例如：“log.#”能够匹配到“log.info.oa”
"log.*"只会匹配到"log.erro"
//生产端代码
//1 创建ConnectionFactory
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/"); //2 创建Connection
Connection connection = connectionFactory.newConnection();
//3 创建Channel
Channel channel = connection.createChannel();
//4 声明
String exchangeName = "test_topic_exchange";
String routingKey1 = "user.save";
String routingKey2 = "user.update";
String routingKey3 = "user.delete.abc";
//5 发送 String msg = "Hello World RabbitMQ 4 Topic Exchange Message ...";
channel.basicPublish(exchangeName, routingKey1 , null , msg.getBytes());
channel.basicPublish(exchangeName, routingKey2 , null , msg.getBytes());
channel.basicPublish(exchangeName, routingKey3 , null , msg.getBytes());
channel.close();
connection.close();
1 //消费端代码
2 ConnectionFactory connectionFactory = new ConnectionFactory() ;
3
4 connectionFactory.setHost("127.0.0.1");
5 connectionFactory.setPort(5672);
6 connectionFactory.setVirtualHost("/");
7
8 connectionFactory.setAutomaticRecoveryEnabled(true);
9 connectionFactory.setNetworkRecoveryInterval(3000);
10 Connection connection = connectionFactory.newConnection();
11
12 Channel channel = connection.createChannel();
13 //4 声明
14 String exchangeName = "test_topic_exchange";
15 String exchangeType = "topic";
16 String queueName = "test_topic_queue";
17
//只能接收到前两条消息
//String routingKey = "user.*"; //可以接收到三条消息
18 String routingKey = "user.#";
19 // 1 声明交换机
20 channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
21 // 2 声明队列
22 channel.queueDeclare(queueName, false, false, false, null);
23 // 3 建立交换机和队列的绑定关系:
24 channel.queueBind(queueName, exchangeName, routingKey);
25
26 //durable 是否持久化消息
27 QueueingConsumer consumer = new QueueingConsumer(channel);
28 //参数：队列名称、是否自动ACK、Consumer
29 channel.basicConsume(queueName, true, consumer);
30 //循环获取消息
31 while(true){
32 //获取消息，如果没有消息，这一步将会一直阻塞
33 Delivery delivery = consumer.nextDelivery();
34 String msg = new String(delivery.getBody());
35 System.out.println("收到消息：" + msg);
36 }
不做任何路由的连接方式：
Fanout Exchange：不处理路由键，只需简单的将队列绑定到交换机上
发送到交换机的消息都会被转发到与该交换机绑定的队列上
fanout交换机转发消息是最快的
//生产者代码
//1 创建ConnectionFactory
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
//2 创建Connection
Connection connection = connectionFactory.newConnection();
//3 创建Channel
Channel channel = connection.createChannel();
//4 声明
String exchangeName = "test_fanout_exchange";
//5 发送
for(int i = 0; i < 10; i ++) {
String msg = "Hello World RabbitMQ 4 FANOUT Exchange Message ...";
channel.basicPublish(exchangeName, "", null , msg.getBytes());
}
channel.close();
connection.close(); //消费者端代码
ConnectionFactory connectionFactory = new ConnectionFactory() ;
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
connectionFactory.setAutomaticRecoveryEnabled(true);
connectionFactory.setNetworkRecoveryInterval(3000);
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
//4 声明
String exchangeName = "test_fanout_exchange";
String exchangeType = "fanout";
String queueName = "test_fanout_queue";
String routingKey = ""; //不设置路由键
channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
//durable 是否持久化消息
QueueingConsumer consumer = new QueueingConsumer(channel);
//参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(queueName, true, consumer);
//循环获取消息
while(true){
//获取消息，如果没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery();
String msg = new String(delivery.getBody());
System.out.println("收到消息：" + msg);
} Headers Exchange：不常用，使用消息头来进行路由的