RabbitMQ完整使用

小鱼儿 2023-02-21 06:12 4阅读 0赞

### RabbitMQ几种模式足够使用了、详细信息点击下面项目地址已经总结好了文档 ###

*   *   *  一、点对点模式(多种案例)
         *   *   *   *  1、点对点单机模式
                     *  2、消费者集群均摊消费
                     *  3、消息应答ACK模式
                     *  3、公平队列模式
         *  二、fanout模式
         *  三、Routing模式
         *  四、Topic模式

项目下载地址： [点我吧-顺便来个赞][-]

<table> 
 <thead> 
  <tr> 
   <th align="center">公众号</th> 
   <th align="right">微服务交流群</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">hxx_service_quickly</td> 
   <td align="right"><img src="https://img-blog.csdnimg.cn/20200701104015397.png" alt="在这里插入图片描述"></td> 
  </tr> 
 </tbody> 
</table>

### 一、点对点模式(多种案例) ###

###### 1、点对点单机模式 ######

生产者

package com.hxx.danji;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * 1、生产者、 */
    public class Producer { 
    
    
    	//队列
    	private static final String QUEUE_NAME = "hxx-test_queue";
    
    	public static void main(String[] args) throws IOException, TimeoutException, TimeoutException { 
    		// 1.获取连接
    		Connection newConnection = MQConnectionUtils.newConnection();
    		// 2.创建通道
    		Channel channel = newConnection.createChannel();
    
    		/** * 3.创建队列声明 * * 参数1、队列名称 * 参数2 是否持久化 */
    		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
    		//消息
    		String msg = "test_hxx-110";
    		System.out.println("生产者发送消息:" + msg);
    		// 4.发送消息
    		channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
    		channel.close();
    		newConnection.close();
    	}
    
    }

消费者

package com.hxx.danji;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * 1、消费者 * 2、单机版 */
    public class Customer { 
        private static final String QUEUE_NAME = "hxx-test_queue";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            System.out.println("002");
            // 1.获取连接
            Connection newConnection = MQConnectionUtils.newConnection();
            // 2.获取通道
            Channel channel = newConnection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                        throws IOException { 
                    String msgString = new String(body, "UTF-8");
                    System.out.println("消费者获取消息:" + msgString);
                }
            };
            // 3.监听队列
            channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
    
        }
    
    }

创建链接的工具类

package com.hxx.utils;
    
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * 1、创建连接 */
    public class MQConnectionUtils { 
    
    	public static Connection newConnection() throws IOException, TimeoutException { 
    		// 1.定义连接工厂
    		ConnectionFactory factory = new ConnectionFactory();
    		// 2.设置服务器地址
    		factory.setHost("127.0.0.1");
    		// 3.设置协议端口号
    		factory.setPort(5672);
    		// 4.设置vhost
    		factory.setVirtualHost("/admin_host");
    		// 5.设置用户名称
    		factory.setUsername("hxx");
    		// 6.设置用户密码
    		factory.setPassword("103658");
    		// 7.创建新的连接
    		Connection newConnection = factory.newConnection();
    		return newConnection;
    	}
    
    }

###### 2、消费者集群均摊消费 ######

生产者

package com.hxx.jiqun;
    
    import com.hxx.danji.MQConnectionUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * 1、消费者集群 均摊消费 * ①、 Customer01 * ② Customer02 */
    public class Producer { 
    
    
    	//队列
    	private static final String QUEUE_NAME = "hxx-test_queue";
    
    	public static void main(String[] args) throws IOException, TimeoutException, TimeoutException { 
    		// 1.获取连接
    		Connection newConnection = MQConnectionUtils.newConnection();
    		// 2.创建通道
    		Channel channel = newConnection.createChannel();
    
    		/** * 3.创建队列声明 * * 参数1、队列名称 * 参数2 是否持久化 */
    		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
    		for (int i = 0; i < 10; i++) { 
    			//消息
    			String msg = "test_hxx-110" +i;
    			System.out.println("生产者投递消息:" + msg );
    			// 4.发送消息
    			channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
    		}
    		channel.close();
    		newConnection.close();
    	}
    
    }

消费者1

package com.hxx.jiqun;
    
    import com.hxx.danji.MQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * 1、消费者集群 * */
    public class Customer01 { 
        private static final String QUEUE_NAME = "hxx-test_queue";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
          // System.out.println("002");
            // 1.获取连接
            Connection newConnection = MQConnectionUtils.newConnection();
            // 2.创建通道
          final   Channel channel = newConnection.createChannel();
    
            //3、消费者关联队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { 
    
              //监听获取到的消息
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                        throws IOException { 
                    String msgString = new String(body, "UTF-8");
    
                    System.out.println("消费者获取消息01:" + msgString);
    
    
                }
            };
    
    
    
            channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
    
        }
    
    }

消费者2

package com.hxx.jiqun;
    
    import com.hxx.danji.MQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * 1、消费者集群 */
    public class Customer02 { 
        private static final String QUEUE_NAME = "hxx-test_queue";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
           // System.out.println("002");
            // 1.获取连接
            Connection newConnection = MQConnectionUtils.newConnection();
            // 2.获取通道
            Channel channel = newConnection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                        throws IOException { 
                    String msgString = new String(body, "UTF-8");
                    System.out.println("消费者获取消息02:" + msgString);
                }
            };
            // 3.监听队列
            channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
    
        }
    
    }

###### 3、消息应答ACK模式 ######

生产者

package com.hxx.yingda;
    
    import com.hxx.danji.MQConnectionUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * 1、应答模式的生产者 * ①、 * ② */
    public class ProducerAck { 
    
    
    	//队列
    	private static final String QUEUE_NAME = "hxx-test_queue";
    
    	public static void main(String[] args) throws IOException, TimeoutException, TimeoutException { 
    		// 1.获取连接
    		Connection newConnection = MQConnectionUtils.newConnection();
    		// 2.创建通道
    		Channel channel = newConnection.createChannel();
    
    		/** * 3.创建队列声明 * * 参数1、队列名称 * 参数2 是否持久化 */
    		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
    		for (int i = 0; i < 10; i++) { 
    			//消息
    			String msg = "test_hxx-110" +i;
    			System.out.println("应答模式生产者投递消息:" + msg );
    			// 4.发送消息
    			channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
    		}
    		channel.close();
    		newConnection.close();
    	}
    
    }

消费者

package com.hxx.yingda;
    
    import com.hxx.danji.MQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * 1、消费者应答模式 * 2、手动签收 */
    public class CustomerAck01 { 
        private static final String QUEUE_NAME = "hxx-test_queue";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
          // System.out.println("002");
            // 1.获取连接
            Connection newConnection = MQConnectionUtils.newConnection();
            // 2.创建通道
          final   Channel channel = newConnection.createChannel();
    
            //3、消费者关联队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { 
    
              //监听获取到的消息
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                        throws IOException { 
                    String msgString = new String(body, "UTF-8");
    
                    System.out.println("消费者获取消息(应答模式):" + msgString);
    
                    //设置手动应答模式、 告诉消息队列服务器 消费成功
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
    
    
            /** * 设置应答模式、 * 如果为true情况下、表示为自动应答模式 * 分为自动应答和手动应答 * * 自动应答：不在乎消费者对这个消息处理是否成功、都会告诉队列删除该消息、 * 如如处理消息失败情况下、实现自动补偿 * * 手动应答：消费处理完业务逻辑、手动返回ack(通知) 、 * 告诉给对列服务器 是否删除该消息 */
            channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
    
        }
    
    }

###### 3、公平队列模式 ######

生产者

package com.hxx.gongping;
    
    import com.hxx.danji.MQConnectionUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * 1、公平队列生产者、 */
    public class Producer { 
    
    
    	//队列
    	private static final String QUEUE_NAME = "hxx-test_queue";
    
    	public static void main(String[] args) throws IOException, TimeoutException, TimeoutException { 
    		// 1.获取连接
    		Connection newConnection = MQConnectionUtils.newConnection();
    		// 2.创建通道
    		Channel channel = newConnection.createChannel();
    
    		/** * 3.创建队列声明 * * 参数1、队列名称 * 参数2 是否持久化 */
    		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
    		for (int i = 0; i < 50; i++) { 
    			//消息
    			String msg = "test_hxx-110" +i;
    			System.out.println("应答模式生产者投递消息:" + msg );
    			// 4.发送消息
    			channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
    		}
    		channel.close();
    		newConnection.close();
    	}
    
    }

消费者1

package com.hxx.gongping;
    
    import com.hxx.danji.MQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * 1、消费者 * 2、单机版 */
    public class Customer01 { 
        private static final String QUEUE_NAME = "hxx-test_queue";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            // 1.获取连接
            Connection newConnection = MQConnectionUtils.newConnection();
            // 2.获取通道
          final   Channel channel = newConnection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            channel.basicQos(1);  //每次消费一条、必须消费完才能继续消费
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { 
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                        throws IOException { 
                    String msgString = new String(body, "UTF-8");
                    System.out.println("公平消费者获取消息:" + msgString);
    
                    try { 
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }finally { 
                          //手动应答模式开启
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                }
            };
            // 3.监听队列
            channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
    
        }
    
    }

消费者2

package com.hxx.gongping;
    
    import com.hxx.danji.MQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * 1、消费者应答模式 * 2、手动签收 */
    public class Customer02 { 
        private static final String QUEUE_NAME = "hxx-test_queue";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            // System.out.println("002");
            // 1.获取连接
            Connection newConnection = MQConnectionUtils.newConnection();
            // 2.创建通道
            final Channel channel = newConnection.createChannel();
    
            //3、消费者关联队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.basicQos(1);  //每次消费一条、必须消费完才能继续消费
            
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { 
                //监听获取到的消息
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                        throws IOException { 
                    String msgString = new String(body, "UTF-8");
    
                    System.out.println("公平消费者获取消息(应答模式):" + msgString);
    
                    try { 
                        Thread.sleep(500);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        //手动应答模式开启
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
    
            /** * 设置应答模式、 * 如果为true情况下、表示为自动应答模式 * 分为自动应答和手动应答 * * 自动应答：不在乎消费者对这个消息处理是否成功、都会告诉队列删除该消息、 * 如如处理消息失败情况下、实现自动补偿 * * 手动应答：消费处理完业务逻辑、手动返回ack(通知) 、 * 告诉给对列服务器 是否删除该消息 */
            channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
    
        }
    
    }

### 二、fanout模式 ###

生产者

package com.hxx.fanout;
    
    import com.hxx.utils.MQConnectionUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @Author 小坏 * @Date 2020/6/29 17:08 * @Version 1.0 * @program: 父工程 RabbitMQ 万物起源之地 */
    
    
    //生产者交换机类型 producerFanout类型
    public class FanoutProducer { 
        //交换机名称
        private static final String DESTINATION_NAME = "my_fanout_estimation";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            //1、创建mq链接
            Connection connection = MQConnectionUtils.newConnection();
    
            //2、创建通道
            Channel channel = connection.createChannel();
    
            /** * 3、生产者绑定交换机 * 参数1、交换机名称 * 参数2 、交换机类型 * * fanout只要绑定交换机、都会收到信息 */
            channel.exchangeDeclare(DESTINATION_NAME, "fanout");
    
            //4、创建消息
            String msg = "my_fanout_estimation_msg";
    
            //5、发送消息my_fanout_estimation
            channel.basicPublish(DESTINATION_NAME, "", null, msg.getBytes());
    
            //6、关闭通道
            channel.close();
            connection.close();
        }
    }

消费者邮件队列

package com.hxx.fanout;
    
    import com.hxx.utils.MQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @Author 小坏 * @Date 2020/6/29 17:54 * @Version 1.0 * @program 邮件消费者 */
    
    
    public class CustomerEmailFanout { 
    
    
        //队列名称
        private static final String EMAIL_QUEUE = "email_queue_fanout";
    
        //交换机名称
        private static final String DESTINATION_NAME = "my_fanout_estimation";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            System.out.println("邮件消费者:");
            //1、创建mq链接
            Connection connection = MQConnectionUtils.newConnection();
            //2、创建通道
            Channel channel = connection.createChannel();
    
            //3、消费声明队列
            channel.queueDeclare(EMAIL_QUEUE, false, false, false, null);
    
    
            //4、消费者绑定队列
            channel.queueBind(EMAIL_QUEUE, DESTINATION_NAME, "");
    
            //5、消费监听消息
            DefaultConsumer defaultconsumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(java.lang.String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String msg = new String(body, "UTF-8");
                    System.out.println("邮件消费者获取生产消息" + msg);
                }
            };
    
            //自动签收
            channel.basicConsume(EMAIL_QUEUE, true, defaultconsumer);
    
        }
    }

消费者短信队列

package com.hxx.fanout;
    
    import com.hxx.utils.MQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @Author 小坏 * @Date 2020/6/29 17:54 * @Version 1.0 * @program 短信消费者 */
    
    
    public class CustomerSmsFanout { 
    
    
        //队列名称
        private static final String SMS_QUEUE = "sms_queue_fanout";
    
        //交换机名称
        private static final String DESTINATION_NAME = "my_fanout_estimation";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            System.out.println("短信消费者:");
            //1、创建mq链接
            Connection connection = MQConnectionUtils.newConnection();
            //2、创建通道
            Channel channel = connection.createChannel();
    
            //3、消费声明队列
            channel.queueDeclare(SMS_QUEUE, false, false, false, null);
    
    
            //4、消费者绑定队列 第三个参数不能给null
            channel.queueBind(SMS_QUEUE, DESTINATION_NAME, "");
    
            //5、消费监听消息
            DefaultConsumer defaultconsumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String msg = new String(body, "UTF-8");
                    System.out.println("短信消费者获取生产消息" + msg);
                }
            };
    
            //自动签收
            channel.basicConsume(SMS_QUEUE, true, defaultconsumer);
    
        }
    }

### 三、Routing模式 ###

生产者路由和交换机绑定根据路由key决定去哪个消费

package hxx.topic;
    
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import hxx.utils.MQConnectionUtils;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @Author 小坏 * @Date 2020/6/29 17:08 * @Version 1.0 * @program: routing 路由模式 */
    
    
    //生产者交换机类型 producerFanout类型
    public class RoutingProducer { 
        //交换机名称
        private static final String DESTINATION_NAME = "my_routing_estimation";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            //1、创建mq链接
            Connection connection = MQConnectionUtils.newConnection();
    
            //2、创建通道
            Channel channel = connection.createChannel();
    
            /** * 3、生产者绑定交换机 * 参数1、交换机名称 * 参数2 、交换机类型 * * fanout只要绑定交换机、都会收到信息 */
            channel.exchangeDeclare(DESTINATION_NAME, "direct");
    
    
            String RoutingKey = "email";
            //4、创建消息
            String msg = "my_routing_estimation_msg" + RoutingKey;
    
            /** * 5、发送消息my_routing_estimation * 参数2 是routingkey 路由键 */
    
            channel.basicPublish(DESTINATION_NAME, RoutingKey, null, msg.getBytes());
    
            //6、关闭通道
            channel.close();
            connection.close();
        }
    }

消费者邮件

package hxx.topic;
    
    
    import com.rabbitmq.client.*;
    import hxx.utils.MQConnectionUtils;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @Author 小坏 * @Date 2020/6/29 17:54 * @Version 1.0 * @program 邮件消费者 */
    
    
    public class CustomerEmailRouting { 
    
    
        //队列名称
        private static final String EMAIL_QUEUE = "email_queue_routing";
    
        //交换机名称
        private static final String DESTINATION_NAME = "my_routing_estimation";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            System.out.println("邮件消费者:");
            //1、创建mq链接
            Connection connection = MQConnectionUtils.newConnection();
            //2、创建通道
            Channel channel = connection.createChannel();
    
            //3、消费声明队列
            channel.queueDeclare(EMAIL_QUEUE, false, false, false, null);
    
    
            //4、消费者绑定队列
            channel.queueBind(EMAIL_QUEUE, DESTINATION_NAME, "email");
            //如果想收到短信既能收到邮件、多谢一个绑定即可
            //channel.queueBind(EMAIL_QUEUE, DESTINATION_NAME, "sms");
    
    
            //5、消费监听消息
            DefaultConsumer defaultconsumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String msg = new String(body, "UTF-8");
                    System.out.println("邮件消费者获取生产消息" + msg);
                }
            };
    
            //自动签收
            channel.basicConsume(EMAIL_QUEUE, true, defaultconsumer);
    
        }
    }

消费者短信

package hxx.topic;
    
    import com.rabbitmq.client.*;
    import hxx.utils.MQConnectionUtils;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @Author 小坏 * @Date 2020/6/29 17:54 * @Version 1.0 * @program 短信消费者 */
    
    
    public class CustomerSmsRouting { 
    
    
        //队列名称
        private static final String SMS_QUEUE = "sms_queue_routing";
    
        //交换机名称
        private static final String DESTINATION_NAME = "my_routing_estimation";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            System.out.println("短信消费者:");
            //1、创建mq链接
            Connection connection = MQConnectionUtils.newConnection();
            //2、创建通道
            Channel channel = connection.createChannel();
    
            //3、消费声明队列
            channel.queueDeclare(SMS_QUEUE, false, false, false, null);
    
    
            //4、消费者绑定队列 第三个参数不能给null
            channel.queueBind(SMS_QUEUE, DESTINATION_NAME, "email");
            channel.queueBind(SMS_QUEUE, DESTINATION_NAME, "sms");
    
            //如果想收到短信既能收到邮件、多谢一个绑定即可
            //channel.queueBind(EMAIL_QUEUE, DESTINATION_NAME, "sms");
    
            //5、消费监听消息
            DefaultConsumer defaultconsumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String msg = new String(body, "UTF-8");
                    System.out.println("短信消费者获取生产消息" + msg);
                }
            };
    
            //自动签收
            channel.basicConsume(SMS_QUEUE, true, defaultconsumer);
    
        }
    }

### 四、Topic模式 ###

生产者

package hxx.topic;
    
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import hxx.utils.MQConnectionUtils;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @Author 小坏 * @Date 2020/6/29 17:08 * @Version 1.0 * @program: topic 路由模式 * * 1、消费者可以通过通配符来决定是否接受消息 * log.* 代表直接收一个 * log.# 全部接受 */
    
    
    //生产者交换机类型 producerFanout类型
    public class TopicProducer { 
        //交换机名称
        private static final String DESTINATION_NAME = "my_topic_estimation";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            //1、创建mq链接
            Connection connection = MQConnectionUtils.newConnection();
    
            //2、创建通道
            Channel channel = connection.createChannel();
    
            /** * 3、生产者绑定交换机 * 参数1、交换机名称 * 参数2 、交换机类型 * * fanout只要绑定交换机、都会收到信息 */
            channel.exchangeDeclare(DESTINATION_NAME, "topic");
    
            String RoutingKey = "log.sms";
            //4、创建消息
            String msg = "my_topic_estimation_msg" + RoutingKey;
    
            /** * 5、发送消息my_routing_estimation * 参数2 是routingkey 路由键 */
    
            channel.basicPublish(DESTINATION_NAME, RoutingKey, null, msg.getBytes());
    
            //6、关闭通道
            channel.close();
            connection.close();
        }
    }

消费者短信

package hxx.topic;
    
    import com.rabbitmq.client.*;
    import hxx.utils.MQConnectionUtils;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @Author 小坏 * @Date 2020/6/29 17:54 * @Version 1.0 * @program topic 路由模式 短信消费者 * * 1、消费者可以通过通配符来决定是否接受消息 * log.* 代表直接收一个 * log.# 全部接受 */
    
    
    public class CustomerSmsTopic { 
    
    
        //队列名称
        private static final String SMS_QUEUE = "sms_queue_topic";
    
        //交换机名称
        private static final String DESTINATION_NAME = "my_topic_estimation";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            System.out.println("短信消费者:");
            //1、创建mq链接
            Connection connection = MQConnectionUtils.newConnection();
            //2、创建通道
            Channel channel = connection.createChannel();
    
            //3、消费声明队列
            channel.queueDeclare(SMS_QUEUE, false, false, false, null);
    
    
            //4、消费者绑定队列 第三个参数不能给null
            channel.queueBind(SMS_QUEUE, DESTINATION_NAME, "log.sms");
    
            //如果想收到短信既能收到邮件、多谢一个绑定即可
            //channel.queueBind(EMAIL_QUEUE, DESTINATION_NAME, "sms");
    
            //5、消费监听消息
            DefaultConsumer defaultconsumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String msg = new String(body, "UTF-8");
                    System.out.println("短信消费者获取生产消息" + msg);
                }
            };
    
            //自动签收
            channel.basicConsume(SMS_QUEUE, true, defaultconsumer);
    
        }
    }

消费者邮件

package hxx.topic;
    
    
    import com.rabbitmq.client.*;
    import hxx.utils.MQConnectionUtils;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @Author 小坏 * @Date 2020/6/29 17:54 * @Version 1.0 * @program topic 路由模式 邮件消费者 * * * 1、消费者可以通过通配符来决定是否接受消息 * log.* 代表直接收一个 * log.# 全部接受 */
    
    public class CustomerEmailTopic { 
        //队列名称
        private static final String EMAIL_QUEUE = "email_queue_topic";
    
        //交换机名称
        private static final String DESTINATION_NAME = "my_topic_estimation";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            System.out.println("邮件消费者:");
            //1、创建mq链接
            Connection connection = MQConnectionUtils.newConnection();
            //2、创建通道
            Channel channel = connection.createChannel();
    
            //3、消费声明队列
            channel.queueDeclare(EMAIL_QUEUE, false, false, false, null);
    
    
            //4、消费者绑定队列
          // channel.queueBind(EMAIL_QUEUE, DESTINATION_NAME, "log.*");
            channel.queueBind(EMAIL_QUEUE, DESTINATION_NAME, "log.*");
            //如果想收到短信既能收到邮件、多谢一个绑定即可
            //channel.queueBind(EMAIL_QUEUE, DESTINATION_NAME, "sms");
    
    
            //5、消费监听消息
            DefaultConsumer defaultconsumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String msg = new String(body, "UTF-8");
                    System.out.println("邮件消费者获取生产消息" + msg);
                }
            };
    
            //自动签收
            channel.basicConsume(EMAIL_QUEUE, true, defaultconsumer);
    
        }
    }

消费者钉钉

package hxx.topic;
    
    import com.rabbitmq.client.*;
    import hxx.utils.MQConnectionUtils;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @Author 小坏 * @Date 2020/6/29 17:54 * @Version 1.0 * @program topic 路由模式 短信消费者 * * 1、消费者可以通过通配符来决定是否接受消息 * log.* 代表直接收一个 * log.# 全部接受 */
    
    
    public class CustomerDingDingTopic { 
    
    
        //队列名称
        private static final String SMS_QUEUE = "dingDing_queue_topic";
    
        //交换机名称
        private static final String DESTINATION_NAME = "my_topic_estimation";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            System.out.println("钉钉消费者:");
            //1、创建mq链接
            Connection connection = MQConnectionUtils.newConnection();
            //2、创建通道
            Channel channel = connection.createChannel();
    
            //3、消费声明队列
            channel.queueDeclare(SMS_QUEUE, false, false, false, null);
    
    
            //4、消费者绑定队列 第三个参数不能给null
            channel.queueBind(SMS_QUEUE, DESTINATION_NAME, "log.#");
    
            //如果想收到短信既能收到邮件、多谢一个绑定即可
            //channel.queueBind(EMAIL_QUEUE, DESTINATION_NAME, "sms");
    
            //5、消费监听消息
            DefaultConsumer defaultconsumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String msg = new String(body, "UTF-8");
                    System.out.println("钉钉消费者获取生产消息" + msg);
                }
            };
    
            //自动签收
            channel.basicConsume(SMS_QUEUE, true, defaultconsumer);
    
        }
    }

[-]: https://gitee.com/xiaohuai09136016/RabbitMq