RabbitMQ六种工作模式有哪些?怎样用SpringBoot整合RabbitMQ

Hadoop2.7.7阿里云安装部署

目录

  • 一、RabbitMQ入门程序
  • 二、Work queues 工作模式
  • 三、Publish / Subscribe 发布/订阅模式
  • 四、Routing 路由模式
  • 五、Topics
  • 六、Header
  • 七、RPC
  • 八、Spring Data Elasticsearch

一、RabbitMQ入门程序

<dependencies> <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> </dependencies> 

application.yml

server: port: 44000 spring: application: name: test-rabbitmq-producer rabbitmq: username: guest password: guest host: 127.0.0.1 port: 5672 virtual-host: / 

消息发送者

/** * Description: rabbitmq入门程序 * * @author zygui * @date Created on 2020/5/13 15:34 */ public class Producer01 { // 声明一个消息队列名称 private static final String QUEUE_NAME = "helloworld"; 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服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的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 是否独占连接,队列只允许在该连接中访问,如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建 * 4、autoDelete 自动删除,队列不再使用时是否自动删除此队列,如果将此参数和exclusive参数设置为true就可以实现临时队列(队列不用了就自动删除) * 5、arguments 参数,可以设置一个队列的扩展参数,比如:可设置存活时间 */ channel.queueDeclare(QUEUE_NAME,true,false,false,null); // 发送消息 //参数:String exchange, String routingKey, BasicProperties props, byte[] body /** * 参数明细: * 1、exchange,交换机,如果不指定将使用mq的默认交换机(设置为"") * 2、routingKey,路由key,交换机根据路由key来将消息转发到指定的队列,如果使用默认交换机,routingKey设置为队列的名称 * 3、props,消息的属性 * 4、body,消息内容 */ //消息内容 String message = "hello world 桂朝阳"; channel.basicPublish("",QUEUE_NAME,null,message.getBytes()); System.out.println("send to mq "+message); } catch (Exception e) { e.printStackTrace(); } finally { try { // 关闭信道 channel.close(); // 关闭连接 connection.close(); } catch (Exception e) { e.printStackTrace(); } } } } 

消息接收者

/** * Description: rabbitmq入门程序 * * @author zygui * @date Created on 2020/5/13 15:45 */ public class Consumer01 { private static final String QUEUE_NAME = "helloworld"; 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服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mq connectionFactory.setVirtualHost("/"); //建立新连接 Connection connection = connectionFactory.newConnection(); //创建会话通道,生产者和mq服务所有通信都在channel通道中完成 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_NAME,true,false,false,null); // 实现消费方法 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { /** * 当接收到消息后此方法将被调用 * @param consumerTag 消费者标签,用来标识消费者的,在监听队列时设置channel.basicConsume * @param envelope 信封,通过envelope * @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("receive message:"+message); } }; //监听队列 //参数:String queue, boolean autoAck, Consumer callback /** * 参数明细: * 1、queue 队列名称 * 2、autoAck 自动回复,当消费者接收到消息后要告诉mq消息已接收,如果将此参数设置为tru表示会自动回复mq,如果设置为false要通过编程实现回复 * 3、callback,消费方法,当消费者接收到消息要执行的方法 */ channel.basicConsume(QUEUE_NAME,true,defaultConsumer); } } 

入门程序步骤

二、Work queues 工作模式


三、Publish / Subscribe 发布/订阅模式


消息生产者

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服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的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 是否独占连接,队列只允许在该连接中访问,如果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<5;i++){ //消息内容 String message = "send inform message to user"; 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(); } } } } 

消息接收者1

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服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mq connectionFactory.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 */ @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("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); } } 

消息接收者2

public class Consumer02_subscribe_sms { //队列名称 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) throws IOException, TimeoutException { //通过连接工厂创建新的连接和mq建立连接 ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("127.0.0.1"); connectionFactory.setPort(5672);//端口 connectionFactory.setUsername("guest"); connectionFactory.setPassword("guest"); //设置虚拟机,一个mq服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mq connectionFactory.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_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_SMS, EXCHANGE_FANOUT_INFORM, ""); //实现消费方法 DefaultConsumer defaultConsumer = new DefaultConsumer(channel){ /** * 当接收到消息后此方法将被调用 * @param consumerTag 消费者标签,用来标识消费者的,在监听队列时设置channel.basicConsume * @param envelope 信封,通过envelope * @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("receive message:"+message); } }; //监听队列 //参数:String queue, boolean autoAck, Consumer callback /** * 参数明细: * 1、queue 队列名称 * 2、autoAck 自动回复,当消费者接收到消息后要告诉mq消息已接收,如果将此参数设置为tru表示会自动回复mq,如果设置为false要通过编程实现回复 * 3、callback,消费方法,当消费者接收到消息要执行的方法 */ channel.basicConsume(QUEUE_INFORM_SMS,true,defaultConsumer); } } 

四、Routing 路由模式


消息生产者

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服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的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 是否独占连接,队列只允许在该连接中访问,如果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++){ //发送消息的时候指定routingKey String 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++){ //发送消息的时候指定routingKey String message = "send sms inform message to user"; channel.basicPublish(EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS,null,message.getBytes()); System.out.println("send to mq "+message); } // 此时指定的路由键是 inform, 所以两个消费者都可以消费 /*for(int i=0;i<5;i++){ //发送消息的时候指定routingKey String 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(); } } } } 

消息接收者1

卷积神经网络学习笔记——SENet

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服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mq connectionFactory.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 */ @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("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); } }
http://www.dtmao.cc/news_show_631033.shtml

消息接收者2

public class Consumer03_routing_sms { //队列名称 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_SMS="inform_sms"; 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服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mq connectionFactory.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_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_SMS, EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS); //实现消费方法 DefaultConsumer defaultConsumer = new DefaultConsumer(channel){ /** * 当接收到消息后此方法将被调用 * @param consumerTag 消费者标签,用来标识消费者的,在监听队列时设置channel.basicConsume * @param envelope 信封,通过envelope * @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("receive message:"+message); } }; //监听队列 //参数:String queue, boolean autoAck, Consumer callback /** * 参数明细: * 1、queue 队列名称 * 2、autoAck 自动回复,当消费者接收到消息后要告诉mq消息已接收,如果将此参数设置为tru表示会自动回复mq,如果设置为false要通过编程实现回复 * 3、callback,消费方法,当消费者接收到消息要执行的方法 */ channel.basicConsume(QUEUE_INFORM_SMS,true,defaultConsumer); } } 

五、Topics

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服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的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 是否独占连接,队列只允许在该连接中访问,如果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++){ //发送消息的时候指定routingKey String 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++){ //发送消息的时候指定routingKey String 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++){ //发送消息的时候指定routingKey String 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(); } } } } 

消息接收者1

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服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mq connectionFactory.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 */ @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("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); } } 

小写接收者2

public class Consumer04_topics_sms { //队列名称 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_SMS="inform.#.sms.#"; 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服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mq connectionFactory.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_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_SMS, EXCHANGE_TOPICS_INFORM,ROUTINGKEY_SMS); //实现消费方法 DefaultConsumer defaultConsumer = new DefaultConsumer(channel){ /** * 当接收到消息后此方法将被调用 * @param consumerTag 消费者标签,用来标识消费者的,在监听队列时设置channel.basicConsume * @param envelope 信封,通过envelope * @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("receive message:"+message); } }; //监听队列 //参数:String queue, boolean autoAck, Consumer callback /** * 参数明细: * 1、queue 队列名称 * 2、autoAck 自动回复,当消费者接收到消息后要告诉mq消息已接收,如果将此参数设置为tru表示会自动回复mq,如果设置为false要通过编程实现回复 * 3、callback,消费方法,当消费者接收到消息要执行的方法 */ channel.basicConsume(QUEUE_INFORM_SMS,true,defaultConsumer); } } 

六、Header

七、RPC

八、Spring Data Elasticsearch

rabbitmq-producer 消息发送者

@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.#"; public static final String ROUTINGKEY_SMS="inform.#.sms.#"; //声明交换机 @Bean(EXCHANGE_TOPICS_INFORM) public Exchange EXCHANGE_TOPICS_INFORM(){ //durable(true) 持久化,mq重启之后交换机还在 return ExchangeBuilder.topicExchange(EXCHANGE_TOPICS_INFORM).durable(true).build(); } // 声明队列 //声明QUEUE_INFORM_EMAIL队列 @Bean(QUEUE_INFORM_EMAIL) public Queue QUEUE_INFORM_EMAIL(){ return new Queue(QUEUE_INFORM_EMAIL); } //声明QUEUE_INFORM_SMS队列 @Bean(QUEUE_INFORM_SMS) public Queue QUEUE_INFORM_SMS(){ return new Queue(QUEUE_INFORM_SMS); } // 绑定交换机和队列 //ROUTINGKEY_EMAIL队列绑定交换机,指定routingKey @Bean public Binding BINDING_QUEUE_INFORM_EMAIL(@Qualifier(QUEUE_INFORM_EMAIL) Queue queue, @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_EMAIL).noargs(); } //ROUTINGKEY_SMS队列绑定交换机,指定routingKey @Bean public Binding BINDING_ROUTINGKEY_SMS(@Qualifier(QUEUE_INFORM_SMS) Queue queue, @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_SMS).noargs(); } } 

rabbitmq-consumer 消息接收者

@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.#"; public static final String ROUTINGKEY_SMS="inform.#.sms.#"; //声明交换机 @Bean(EXCHANGE_TOPICS_INFORM) public Exchange EXCHANGE_TOPICS_INFORM(){ //durable(true) 持久化,mq重启之后交换机还在 return ExchangeBuilder.topicExchange(EXCHANGE_TOPICS_INFORM).durable(true).build(); } // 声明队列 //声明QUEUE_INFORM_EMAIL队列 @Bean(QUEUE_INFORM_EMAIL) public Queue QUEUE_INFORM_EMAIL(){ return new Queue(QUEUE_INFORM_EMAIL); } //声明QUEUE_INFORM_SMS队列 @Bean(QUEUE_INFORM_SMS) public Queue QUEUE_INFORM_SMS(){ return new Queue(QUEUE_INFORM_SMS); } // 绑定交换机和队列 //ROUTINGKEY_EMAIL队列绑定交换机,指定routingKey @Bean public Binding BINDING_QUEUE_INFORM_EMAIL(@Qualifier(QUEUE_INFORM_EMAIL) Queue queue, @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_EMAIL).noargs(); } //ROUTINGKEY_SMS队列绑定交换机,指定routingKey @Bean public Binding BINDING_ROUTINGKEY_SMS(@Qualifier(QUEUE_INFORM_SMS) Queue queue, @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_SMS).noargs(); } } 
@SpringBootApplication @EnableRabbit public class TestRabbitMQApplication { public static void main(String[] args) { SpringApplication.run(TestRabbitMQApplication.class, args); } } 

监听消息队列

@Component public class ReceiveHandler { @RabbitListener(queues = {RabbitMQConfig.QUEUE_INFORM_EMAIL}) public void receiveMsg(String msg) { System.out.println("接收到的消息是 = " + msg); } } 

在rabbitmq-provider中测试

@SpringBootTest @RunWith(SpringRunner.class) public class Producer05_topics_springboot { // 使用rabbitTemplate发送消息 @Autowired private RabbitTemplate rabbitTemplate; @Test public void testSendEmail() { String message = "send email message to user"; /** * arg1: 交换机名称 * arg2: 路由键 * arg3: 消息内容 */ rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_TOPICS_INFORM, "inform.email", message); } }

CodeMonkey少儿编程第3章 times循环

相关推荐

发表评论

路人甲

网友评论(0)