RabbitMQ整合SpringCloud

交换机

RabbitMQ整合SpringCloud

交换机属性：

Name：交换机名称
Type：交换机类型 direct、topic、fanout、headers
Durability：是否需要持久化，true为持久化
Auto Delete：当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange
Internal：当前Exchange是否用于RabbitMQ内部使用，默认为False
Arguments：扩展参数，用于扩展AMQP协议，定制化使用

直流交换机

直连交换机Direct Exchange（完全匹配路由key）
所有发送到Direct Exchange的消息会被转发到RouteKey中指定的Queue
注意：Direct模式可以使用RabbitMQ自带的Exchange：default Exchange，所以不需要将Exchange进行任何绑定（binding）操作，消息传递时，RouteKey必须完全匹配才会被队列接收，否则该消息会被抛弃；

消费端代码

package com.liuxia.rabbitmqapi.exchange.direct;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;public class Consumer4DirectExchange {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory() ;connectionFactory.setHost("192.168.137.129");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、Consumerchannel.basicConsume(queueName, true, consumer);//循环获取消息while(true){//获取消息，如果没有消息，这一步将会一直阻塞QueueingConsumer.Delivery delivery = consumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println("收到消息：" + msg);}}
}

生产端代码

package com.liuxia.rabbitmqapi.exchange.direct;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer4DirectExchange {public static void main(String[] args) throws Exception {//1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.137.129");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel channel = connection.createChannel();//4 声明String exchangeName = "test_direct_exchange";String routingKey = "test.direct";
//        String routingKey = "test.direct111"; //收不到//5 发送String msg = "Hello World RabbitMQ 4  Direct Exchange Message 111 ... ";channel.basicPublish(exchangeName, routingKey , null , msg.getBytes());}
}

主题交换机

主题交换机Topic Exchange（匹配路由规则的交换机）
所有发送到Topic Exchange的消息被转发到所有关系RouteKey中指定Topic的Queue上；
Exchange将RouteKey和某Topic进行模糊匹配，此时队列需要绑定一个Topic；
注意：可以使用通配符进行模糊匹配
符号：“#” 匹配一个或者多个词
符号：“” 匹配不多不少一个词
列如：
“log.#” 能够匹配到 “log.info.oa”
“log.” 能够匹配到 “log.err”

消费端代码

package com.liuxia.rabbitmqapi.exchange.topic;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;public class Consumer4TopicExchange {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory() ;connectionFactory.setHost("192.168.137.129");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setAutomaticRecoveryEnabled(true);connectionFactory.setNetworkRecoveryInterval(3000);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//4 声明String exchangeName = "test_topic_exchange";String exchangeType = "topic";String queueName = "test_topic_queue";String routingKey = "user.#";
//        String routingKey = "user.*";// 1 声明交换机channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);// 2 声明队列channel.queueDeclare(queueName, false, false, false, null);// 3 建立交换机和队列的绑定关系:channel.queueBind(queueName, exchangeName, routingKey);//durable 是否持久化消息QueueingConsumer consumer = new QueueingConsumer(channel);//参数：队列名称、是否自动ACK、Consumerchannel.basicConsume(queueName, true, consumer);//循环获取消息while(true){//获取消息，如果没有消息，这一步将会一直阻塞QueueingConsumer.Delivery delivery = consumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println("收到消息：" + msg);}}
}

生产端代码

package com.liuxia.rabbitmqapi.exchange.topic;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer4TopicExchange {public static void main(String[] args) throws Exception {//1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.137.129");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel 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();}
}

输出交换机

输出交换机Fanout Exchange（不做路由）
不处理路由键，只需要简单的将队列绑定到交换机上；
发送到交换机的消息都会被转发到与该交换机绑定的所有队列上；
Fanout交换机转发消息是最快的

消费端代码

package com.liuxia.rabbitmqapi.exchange.fanout;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;public class Consumer4FanoutExchange {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory() ;connectionFactory.setHost("192.168.137.129");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、Consumerchannel.basicConsume(queueName, true, consumer);//循环获取消息while(true){//获取消息，如果没有消息，这一步将会一直阻塞QueueingConsumer.Delivery delivery = consumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println("收到消息：" + msg);}}
}

生产端代码

package com.liuxia.rabbitmqapi.exchange.fanout;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer4FanoutExchange {public static void main(String[] args) throws Exception {//1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.137.129");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel 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();}
}