RabbitMQ七种工作模式实现测试代码

骑猪看日落 2022-12-18 03:58 73阅读 0赞

#### 所有工作模式依赖都相同 ####

<dependencies>
            
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.6.0</version>
            </dependency>
        </dependencies>
    
    
        <build>
            <plugins>
                
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.8.0</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                    </configuration>
                </plugin>
            </plugins>

#### 一.简单工作模式代码 ####

和RabbitMQ生产者消费者实列种的代码相同

生产者：

package cn.twgfs.rabbitmq.provider;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author willian * @date 2020/10/13 11:27 */
    public class ProducerHelloWorld { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
             connectionFactory.setHost("192.168.0.103");//设置注解
             connectionFactory.setPort(5672);//设置端口
             connectionFactory.setVirtualHost("/vm");//设置虚拟及
             connectionFactory.setUsername("tang");
             connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
            //5 创建队列Queen
            /**参数介绍： * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) * queue：队列名称 durable：是否持久化 exclusive是否独占，只能有一个消费者监听队列。 * queue：队列名称 durable：是否持久化 exclusive是否独占，只能有一个消费者监听队列。 * autoDelete: 当没有consumer时候是否删除队列 arguments：配置的基本参数 */
            channel.queueDeclare("hello_world",true,false,false,null);
            //6 发送消息到队列
            /**参数介绍： * basicPublish(String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body) *exchange： 交换机的名称，简单模式下交换机会使用默认的 routingKey：路由名称,如果使用默认交换机要和队列名相同 *props 配置信息 body：真是发送的消息数据 */
            String body = "hello_rabbitmq 发送消息";
            channel.basicPublish("","hello_world",null,body.getBytes());
    
            //7 释放连接资源
            channel.close();
            connection.close();
        }
    }

消费者：

package cn.twgfs.rabbitmq.consumer;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author willian * @date 2020/10/13 14:17 */
    public class ConsumerHelloWorld { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
            connectionFactory.setHost("192.168.0.103");//设置注解
            connectionFactory.setPort(5672);//设置端口
            connectionFactory.setVirtualHost("/vm");//设置虚拟及
            connectionFactory.setUsername("tang");
            connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
            //5 创建队列Queen
            /**参数介绍： * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) * queue：队列名称 durable：是否持久化 exclusive是否独占，只能有一个消费者监听队列。 * queue：队列名称 durable：是否持久化 exclusive是否独占，只能有一个消费者监听队列。 * autoDelete: 当没有consumer时候是否删除队列 arguments：配置的基本参数 */
            channel.queueDeclare("hello_world",true,false,false,null);
            //6 从消息队列中消费
            /**参数介绍： *basicConsume(String queue, boolean autoAck, Consumer callback) * queue: 消费的队列名称 autoAck：是否自动消费确认，收到消息确认 * callback：回调对象 */
            Consumer consumer = new DefaultConsumer(channel){ 
                //这是一个回调方法，当收到消息后会自动执行该方法
                /** * @param consumerTag ：标识 * @param envelope： 获取交换机，路由等信息 * @param properties：配置信息 * @param body：真是的数据 * @throws IOException */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println(""+consumerTag);
                    System.out.println("getExchange+"+envelope.getExchange());
                    System.out.println("getRoutingKey+"+envelope.getRoutingKey());
                    System.out.println("properties"+properties);
                    System.out.println("body"+body);
                }
            };
            channel.basicConsume("hello_world",true,consumer);
    
            //7 消费者不关闭连接
        }
    }

#### 二.工作队列模式代码 ####

创建两个类名不同但是，但是代码完全相同的两个消费者  
WorkQueens1，WorkQueens1

package cn.twgfs.rabbitmq.consumer;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author willian * @date 2020/10/13 14:17 */
    public class WorkQueens1 { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
            connectionFactory.setHost("192.168.0.103");//设置注解
            connectionFactory.setPort(5672);//设置端口
            connectionFactory.setVirtualHost("/vm");//设置虚拟及
            connectionFactory.setUsername("tang");
            connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
            //5 创建队列Queen
            /**参数介绍： * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) * queue：队列名称 durable：是否持久化 exclusive是否独占，只能有一个消费者监听队列。 * queue：队列名称 durable：是否持久化 exclusive是否独占，只能有一个消费者监听队列。 * autoDelete: 当没有consumer时候是否删除队列 arguments：配置的基本参数 */
            channel.queueDeclare("work_queens",true,false,false,null);
            //6 从消息队列中消费
            /**参数介绍： *basicConsume(String queue, boolean autoAck, Consumer callback) * queue: 消费的队列名称 autoAck：是否自动消费确认，收到消息确认 * callback：回调对象 */
            Consumer consumer = new DefaultConsumer(channel){ 
                //这是一个回调方法，当收到消息后会自动执行该方法
                /** * @param consumerTag ：标识 * @param envelope： 获取交换机，路由等信息 * @param properties：配置信息 * @param body：真是的数据 * @throws IOException */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("body"+body);
                }
            };
            channel.basicConsume("work_queens",true,consumer);
    
            //7 消费者不关闭连接
        }
    }

生产者代码如下WorkQueens

package cn.twgfs.rabbitmq.provider;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author willian * @date 2020/10/13 11:27 */
    public class WorkQueens { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
             connectionFactory.setHost("192.168.0.103");//设置注解
             connectionFactory.setPort(5672);//设置端口
             connectionFactory.setVirtualHost("/vm");//设置虚拟及
             connectionFactory.setUsername("tang");
             connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
            //5 创建队列Queen
            /**参数介绍： * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) * queue：队列名称 durable：是否持久化 exclusive是否独占，只能有一个消费者监听队列。 * queue：队列名称 durable：是否持久化 exclusive是否独占，只能有一个消费者监听队列。 * autoDelete: 当没有consumer时候是否删除队列 arguments：配置的基本参数 */
            channel.queueDeclare("workqueen_model",true,false,false,null);
            //6 发送消息到队列
            /**参数介绍： * basicPublish(String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body) *exchange： 交换机的名称，简单模式下交换机会使用默认的 routingKey：路由名称,如果使用默认交换机要和队列名相同 *props 配置信息 body：真是发送的消息数据 */
            for (int i = 0; i < 10; i++) { 
                String body = i+"hello_rabbitmq 发送消息";
                channel.basicPublish("","workqueen_model",null,body.getBytes());
            }
            //7 释放连接资源
            channel.close();
            connection.close();
        }
    }

#### 三.Pub/Sub发布订阅模式代码 ####

生产者类：ProducerPubSub

public class ProducerPubSub { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
             connectionFactory.setHost("192.168.0.103");//设置注解
             connectionFactory.setPort(5672);//设置端口
             connectionFactory.setVirtualHost("/vm");//设置虚拟及
             connectionFactory.setUsername("tang");
             connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
    
            //5.创建交换机
            /**参数详细解释 * exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments) * exchange：交换机名称 type：交换机的类型 DIRECT("direct"),定向 FANOUT("fanout"),扇形 TOPIC("topic"),通配符的方式 HEADERS("headers");参数匹配 *durable： 是否持久化 autoDelete： 自动删除 internal：内部使用，一般都是false *arguments: 参数 * durable： 是否持久化 */
            String exchange = "test_exechange";
            channel.exchangeDeclare(exchange, BuiltinExchangeType.FANOUT,true,false,false,null);
            //6.创建两个队列
            String queen1Name = "test_fanout_queen1";
            String queen2Name = "test_fanout_queen2";
            channel.queueDeclare(queen1Name,true,false,false,null);
            channel.queueDeclare(queen2Name,true,false,false,null);
            //7.绑定队列和交换机
            /**参数详解 * queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments) * queue：队列名称 exchange：交换机名称 routingKey：路由键，绑定规则：如果交换机类型为FANOUT，routingKey设置为”“ */
            channel.queueBind(queen1Name,exchange,"");
            channel.queueBind(queen2Name,exchange,"");
            //8.发送消息
            String body = "日志记录：方法";
            channel.basicPublish(exchange,"",null,body.getBytes());
            //9.释放资源
            channel.close();
            connection.close();
    
        }
    }

消费者1:PubSubModel1就是将前面的频道更改为现在带有交换机的频道即可

public class PubSubModel1 { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
            connectionFactory.setHost("192.168.0.103");//设置注解
            connectionFactory.setPort(5672);//设置端口
            connectionFactory.setVirtualHost("/vm");//设置虚拟及
            connectionFactory.setUsername("tang");
            connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
    
            //6 从消息队列中消费
            /**参数介绍： *basicConsume(String queue, boolean autoAck, Consumer callback) * queue: 消费的队列名称 autoAck：是否自动消费确认，收到消息确认 * callback：回调对象 */
            Consumer consumer = new DefaultConsumer(channel){ 
                //这是一个回调方法，当收到消息后会自动执行该方法
                /** * @param consumerTag ：标识 * @param envelope： 获取交换机，路由等信息 * @param properties：配置信息 * @param body：真是的数据 * @throws IOException */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("body"+body);
                    System.out.println("将信息打印到控制台");
                }
            };
            String queen1Name = "test_fanout_queen1";
            channel.basicConsume(queen1Name,true,consumer);
            //7 消费者不关闭连接
        }
    }

消费者2:PubSubModel2就是将前面的频道更改为现在带有交换机的频道即可

package cn.twgfs.rabbitmq.consumer;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author willian * @date 2020/10/13 14:17 */
    public class PubSubModel2 { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
            connectionFactory.setHost("192.168.0.103");//设置注解
            connectionFactory.setPort(5672);//设置端口
            connectionFactory.setVirtualHost("/vm");//设置虚拟及
            connectionFactory.setUsername("tang");
            connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
    
            //6 从消息队列中消费
            /**参数介绍： *basicConsume(String queue, boolean autoAck, Consumer callback) * queue: 消费的队列名称 autoAck：是否自动消费确认，收到消息确认 * callback：回调对象 */
            Consumer consumer = new DefaultConsumer(channel){ 
                //这是一个回调方法，当收到消息后会自动执行该方法
                /** * @param consumerTag ：标识 * @param envelope： 获取交换机，路由等信息 * @param properties：配置信息 * @param body：真是的数据 * @throws IOException */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("body"+body);
                    System.out.println("将信息存储到数据库");
                }
            };
            String queen2Name = "test_fanout_queen2";
            channel.basicConsume(queen2Name,true,consumer);
            //7 消费者不关闭连接
        }
    }

#### 四.Routing路由模式 ####

生产者

package cn.twgfs.rabbitmq.provider;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author willian * @date 2020/10/13 11:27 */
    public class ProducerRouting { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
             connectionFactory.setHost("192.168.0.103");//设置注解
             connectionFactory.setPort(5672);//设置端口
             connectionFactory.setVirtualHost("/vm");//设置虚拟及
             connectionFactory.setUsername("tang");
             connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
    
            //5.创建交换机
            /**参数详细解释 * exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments) * exchange：交换机名称 type：交换机的类型 DIRECT("direct"),定向 FANOUT("fanout"),扇形 TOPIC("topic"),通配符的方式 HEADERS("headers");参数匹配 *durable： 是否持久化 autoDelete： 自动删除 internal：内部使用，一般都是false *arguments: 参数 * durable： 是否持久化 */
            String exchange = "test_direct";
            channel.exchangeDeclare(exchange, BuiltinExchangeType.DIRECT,true,false,false,null);
            //6.创建两个队列
            String queen1Name = "test_direct_queen1";
            String queen2Name = "test_direct_queen2";
            channel.queueDeclare(queen1Name,true,false,false,null);
            channel.queueDeclare(queen2Name,true,false,false,null);
            //7.绑定队列和交换机
            /**参数详解 * queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments) * queue：队列名称 exchange：交换机名称 routingKey：路由键，绑定规则：如果交换机类型为FANOUT，routingKey设置为”“ */
            //1.队列1绑定
            channel.queueBind(queen1Name,exchange,"error");
            //2.队列2的三个绑定
            channel.queueBind(queen2Name,exchange,"info");
            channel.queueBind(queen2Name,exchange,"error");
            channel.queueBind(queen2Name,exchange,"warning");
            //8.发送消息，这里需要携带routingkey后面交换机转发的时候才知道发送到那个队列s1就是路由键
            String body = "日志记录：方法";
            channel.basicPublish(exchange,"info",null,body.getBytes());
            //9.释放资源
            channel.close();
            connection.close();
    
        }
    }

消费者1

package cn.twgfs.rabbitmq.consumer;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author willian * @date 2020/10/13 14:17 */
    public class Routing1 { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
            connectionFactory.setHost("192.168.0.103");//设置注解
            connectionFactory.setPort(5672);//设置端口
            connectionFactory.setVirtualHost("/vm");//设置虚拟及
            connectionFactory.setUsername("tang");
            connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
    
            //6 从消息队列中消费
            /**参数介绍： *basicConsume(String queue, boolean autoAck, Consumer callback) * queue: 消费的队列名称 autoAck：是否自动消费确认，收到消息确认 * callback：回调对象 */
            Consumer consumer = new DefaultConsumer(channel){ 
                //这是一个回调方法，当收到消息后会自动执行该方法
                /** * @param consumerTag ：标识 * @param envelope： 获取交换机，路由等信息 * @param properties：配置信息 * @param body：真是的数据 * @throws IOException */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("body"+body);
                    System.out.println("将信息打印到控制台");
                }
            };
            String queen1Name = "test_direct_queen1";
            channel.basicConsume(queen1Name,true,consumer);
            //7 消费者不关闭连接
        }
    }

消费者2

package cn.twgfs.rabbitmq.consumer;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author willian * @date 2020/10/13 14:17 */
    public class Routing2 { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
            connectionFactory.setHost("192.168.0.103");//设置注解
            connectionFactory.setPort(5672);//设置端口
            connectionFactory.setVirtualHost("/vm");//设置虚拟及
            connectionFactory.setUsername("tang");
            connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
    
            //6 从消息队列中消费
            /**参数介绍： *basicConsume(String queue, boolean autoAck, Consumer callback) * queue: 消费的队列名称 autoAck：是否自动消费确认，收到消息确认 * callback：回调对象 */
            Consumer consumer = new DefaultConsumer(channel){ 
                //这是一个回调方法，当收到消息后会自动执行该方法
                /** * @param consumerTag ：标识 * @param envelope： 获取交换机，路由等信息 * @param properties：配置信息 * @param body：真是的数据 * @throws IOException */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("body"+body);
                    System.out.println("将信息存储到数据库");
                }
            };
            String queen1Name = "test_direct_queen2";
            channel.basicConsume(queen1Name,true,consumer);
            //7 消费者不关闭连接
        }
    }

#### Topic模式 ####

生产者

package cn.twgfs.rabbitmq.provider;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author willian * @date 2020/10/13 11:27 */
    public class ProducerTopic { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
             connectionFactory.setHost("192.168.0.103");//设置注解
             connectionFactory.setPort(5672);//设置端口
             connectionFactory.setVirtualHost("/vm");//设置虚拟及
             connectionFactory.setUsername("tang");
             connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
    
            //5.创建交换机
            /**参数详细解释 * exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments) * exchange：交换机名称 type：交换机的类型 DIRECT("direct"),定向 FANOUT("fanout"),扇形 TOPIC("topic"),通配符的方式 HEADERS("headers");参数匹配 *durable： 是否持久化 autoDelete： 自动删除 internal：内部使用，一般都是false *arguments: 参数 * durable： 是否持久化 */
            String exchange = "test_topic";
            channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC,true,false,false,null);
            //6.创建两个队列
            String queen1Name = "test_topic_queen1";
            String queen2Name = "test_topic_queen2";
            channel.queueDeclare(queen1Name,true,false,false,null);
            channel.queueDeclare(queen2Name,true,false,false,null);
            //7.绑定队列和交换机
            /**参数详解 * queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments) * queue：队列名称 exchange：交换机名称 routingKey：路由键，绑定规则：如果交换机类型为FANOUT，routingKey设置为”“ */
    
            //routing key由两部分组成 系统名称 日志级别
            //需求：所有error级别存数据库 order级别打印到控制台
            channel.queueBind(queen1Name,exchange,"#.error");
            channel.queueBind(queen1Name,exchange,"order.*");
            channel.queueBind(queen2Name,exchange,"*.*");
            //8.发送消息
            String body = "日志记录：方法";
            channel.basicPublish(exchange,"order.info",null,body.getBytes());
            //9.释放资源
            channel.close();
            connection.close();
    
        }
    }

消费者1

package cn.twgfs.rabbitmq.consumer;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author willian * @date 2020/10/13 14:17 */
    public class Topic1 { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
            connectionFactory.setHost("192.168.0.103");//设置注解
            connectionFactory.setPort(5672);//设置端口
            connectionFactory.setVirtualHost("/vm");//设置虚拟及
            connectionFactory.setUsername("tang");
            connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
    
            //6 从消息队列中消费
            /**参数介绍： *basicConsume(String queue, boolean autoAck, Consumer callback) * queue: 消费的队列名称 autoAck：是否自动消费确认，收到消息确认 * callback：回调对象 */
            Consumer consumer = new DefaultConsumer(channel){ 
                //这是一个回调方法，当收到消息后会自动执行该方法
                /** * @param consumerTag ：标识 * @param envelope： 获取交换机，路由等信息 * @param properties：配置信息 * @param body：真是的数据 * @throws IOException */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("body"+body);
                    System.out.println("将信息存储到数据库");
                }
            };
            String queen1Name = "test_topic_queen1";
            channel.basicConsume(queen1Name,true,consumer);
            //7 消费者不关闭连接
        }
    }

消费者2

package cn.twgfs.rabbitmq.consumer;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author willian * @date 2020/10/13 14:17 */
    public class Topic2 { 
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            //1 创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //2 设置参数
            connectionFactory.setHost("192.168.0.103");//设置注解
            connectionFactory.setPort(5672);//设置端口
            connectionFactory.setVirtualHost("/vm");//设置虚拟及
            connectionFactory.setUsername("tang");
            connectionFactory.setPassword("123");
            //3 创建连接
            Connection connection = null;
            connection = connectionFactory.newConnection();
            //4 创建channel
            Channel channel = null;
            channel = connection.createChannel();
    
            //6 从消息队列中消费
            /**参数介绍： *basicConsume(String queue, boolean autoAck, Consumer callback) * queue: 消费的队列名称 autoAck：是否自动消费确认，收到消息确认 * callback：回调对象 */
            Consumer consumer = new DefaultConsumer(channel){ 
                //这是一个回调方法，当收到消息后会自动执行该方法
                /** * @param consumerTag ：标识 * @param envelope： 获取交换机，路由等信息 * @param properties：配置信息 * @param body：真是的数据 * @throws IOException */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("body"+body);
                    System.out.println("将信息打印到控制台");
                }
            };
            String queen2Name = "test_topic_queen2";
            channel.basicConsume(queen2Name,true,consumer);
            //7 消费者不关闭连接
        }
    }