1. 不使用Exchange交换机（默认交换机）
工具类
1package com.lemon.rabbitmq.utils; 2 3import com.rabbitmq.client.Connection; 4import com.rabbitmq.client.ConnectionFactory; 5 6/** 7 * rabbit mq 工具类 ： 获取连接 8 */ 9public class ConnectionUtil { 10 public static Connection getConnection() throws Exception { 11 //创建连接工厂 12 ConnectionFactory connectionFactory = new ConnectionFactory(); 13 14 //主机地址;默认为 localhost 15 connectionFactory.setHost("127.0.0.1"); 16 17 //连接端口;默认为 5672 18 connectionFactory.setPort(5672); 19 20 //虚拟主机名称;默认为 / 21 connectionFactory.setVirtualHost("/lemon"); 22 23 //连接用户名；默认为guest 24 connectionFactory.setUsername("lemon"); 25 26 //连接密码；默认为guest 27 connectionFactory.setPassword("lemon"); 28 29 //创建连接 30 Connection connection = connectionFactory.newConnection(); 31 return connection; 32 } 33} 34 35package cn.lemon.rabbitmq.utils; 36 37import com.rabbitmq.client.AMQP; 38import com.rabbitmq.client.Channel; 39import com.rabbitmq.client.DefaultConsumer; 40import com.rabbitmq.client.Envelope; 41 42import java.io.IOException; 43 44/** 45 * 消费者工具类：生产消息消费者，方便调用 46 */ 47public class ConsumerUtil { 48 49 public static DefaultConsumer getConsumer(Channel channel) { 50 51 DefaultConsumer consumer = new DefaultConsumer(channel) { 52 /** 53 * consumerTag 消费者标签，在channel.basicConsume时候可以指定 54 * envelope 消息包的内容，可从中获取消息id，路由key，交换机等信息 55 * properties 消息属性信息 56 * body 消息内容 57 */ 58 @Override 59 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 60 61 //消息id 62 System.out.println("消息id: " + envelope.getDeliveryTag()); 63 //交换机 64 System.out.println("交换机: " + envelope.getExchange()); 65 //路由key 66 System.out.println("路由key: " + envelope.getRoutingKey()); 67 //接受到的消息 68 System.out.println("收到的消息: " + new String(body, "utf-8")); 69 70 System.out.println("---------------------------------"); 71 72 } 73 }; 74 75 return consumer; 76 } 77}
a. simple简单模式：一个生产者发送消息到队列中由一个消费者接收。
1package cn.lemon.rabbitmq.simple; 2 3import cn.lemon.rabbitmq.utils.ConnectionUtil; 4import com.rabbitmq.client.Channel; 5import com.rabbitmq.client.Connection; 6 7/* 8 消息模式，简单模式: 一个生产者、一个消费者，不需要设置交换机 9 */ 10public class Producer { 11 12 public static final String QUEUE_NAME = "simple_queue"; 13 14 public static void main(String[] args) throws Exception { 15 16 //1. 获取连接 17 Connection connection = ConnectionUtil.getConnection(); 18 19 //2. 创建频道 20 Channel channel = connection.createChannel(); 21 22 /** 23 * 3. 声明（创建）队列 24 * 参数1：队列名称 25 * 参数2：是否定义持久化队列 26 * 参数3：是否独占本次连接 27 * 参数4：是否在不使用的时候自动删除队列 28 * 参数5：队列其它参数 29 */ 30 channel.queueDeclare(QUEUE_NAME, true, false, false, null); 31 32 //4. 发送消息 33 String message = "你好，小兔子"; 34 /** 35 * 参数1：交换机名称，如果没有指定则使用默认Default Exchange 36 * 参数2：路由key,简单模式可以传递队列名称 37 * 参数3：消息其它属性 38 * 参数4：消息内容 39 */ 40 channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); 41 System.out.println("已发送消息：" + message); 42 43 // 5. 关闭资源 44 channel.close(); 45 connection.close(); 46 } 47 48} 49 50package cn.lemon.rabbitmq.simple; 51 52 53import cn.lemon.rabbitmq.utils.ConnectionUtil; 54import cn.lemon.rabbitmq.utils.ConsumerUtil; 55import com.rabbitmq.client.Channel; 56import com.rabbitmq.client.Connection; 57import com.rabbitmq.client.DefaultConsumer; 58 59//消息消费者 60public class Consumer { 61 62 public static void main(String[] args) throws Exception { 63 64 //1. 调用工具类，获取连接 65 Connection connection = ConnectionUtil.getConnection(); 66 67 //2. 创建频道 68 Channel channel = connection.createChannel(); 69 70 //3. 声明队列 这里可以不用申明队列，因为生产者哪里已经创建了 71 // channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null); 72 73 //4. 调用工具类，获取消费者，消费队列中的消息 74 DefaultConsumer consumer = ConsumerUtil.getConsumer(channel); 75 76 //监听消息 77 /** 78 * 参数1：队列名称 79 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了， 80 * mq接收到回复会删除消息，设置为false则需要手动确认 81 * 参数3：消息接收到后回调 82 */ 83 channel.basicConsume(Producer.QUEUE_NAME, true, consumer); 84 85 //不关闭资源，应该一直监听消息 86 87 } 88 89}
b. work工作队列模式：一个生产者发送消息到队列中可由多个消费者接收；多个消费者之间消息是竞争接收。
1package cn.lemon.rabbitmq.work; 2 3import cn.lemon.rabbitmq.utils.ConnectionUtil; 4import com.rabbitmq.client.Channel; 5import com.rabbitmq.client.Connection; 6 7/* 8 消息生产者 : 发送30个消息到队列 9 10 创建两个消费者去监听同一个队列，查看两个消费者接收到的消息是否存在重复。 11 */ 12public class Producer { 13 14 public static final String QUEUE_NAME = "simple_queue"; 15 16 public static void main(String[] args) throws Exception { 17 //1. 创建连接 18 Connection connection = ConnectionUtil.getConnection(); 19 //2. 创建频道 20 Channel channel = connection.createChannel(); 21 //3. 声明队列 22 channel.queueDeclare(QUEUE_NAME, true, false, false, null); 23 24 //4. 发送消息 25 for (int i = 1; i <= 30; i++) { 26 String message = "你好，小兔子! " + i; 27 channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); 28 System.out.println("已发送消息: " + message); 29 } 30 //5. 关闭资源 31 channel.close(); 32 connection.close(); 33 } 34 35} 36 37package cn.itcast.rabbitmq.work; 38 39import cn.lemon.rabbitmq.utils.ConnectionUtil; 40import cn.lemon.rabbitmq.utils.ConsumerUtil; 41import com.rabbitmq.client.Channel; 42import com.rabbitmq.client.Connection; 43import com.rabbitmq.client.DefaultConsumer; 44 45 46//消息消费者 47public class Consumer1 { 48 49 public static void main(String[] args) throws Exception { 50 //1. 创建连接 51 Connection connection = ConnectionUtil.getConnection(); 52 53 //2. 创建频道 54 Channel channel = connection.createChannel(); 55 56 //3. 声明队列 队列已经存在，可以不用创建 57 //channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null); 58 59 //4. 创建消息消费者 60 DefaultConsumer consumer = ConsumerUtil.getConsumer(channel); 61 62 //监听消息 63 /** 64 * 参数1：队列名称 65 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了， 66 * mq接收到回复会删除消息，设置为false则需要手动确认 67 * 参数3：消息接收到后回调 68 */ 69 channel.basicConsume(Producer.QUEUE_NAME, true, consumer); 70 71 //不关闭资源，应该一直监听消息 72 73 } 74 75} 76 77package cn.lemon.rabbitmq.work; 78 79 80import cn.lemon.rabbitmq.utils.ConnectionUtil; 81import cn.lemon.rabbitmq.utils.ConsumerUtil; 82import com.rabbitmq.client.Channel; 83import com.rabbitmq.client.Connection; 84import com.rabbitmq.client.DefaultConsumer; 85 86//消息消费者 87public class Consumer2 { 88 89 public static void main(String[] args) throws Exception { 90 91 //1. 创建连接 92 Connection connection = ConnectionUtil.getConnection(); 93 94 //2. 创建频道 95 Channel channel = connection.createChannel(); 96 97 //3. 声明队列 队列已经存在，不用在创建 98 //channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null); 99 100 //4. 创建消息消费者 101 DefaultConsumer consumer = ConsumerUtil.getConsumer(channel); 102 103 //监听消息 104 channel.basicConsume(Producer.QUEUE_NAME, true, consumer); 105 106 //不关闭资源，应该一直监听消息 107 108 109 } 110 111}
2. 使用Exchange交换机；订阅模式（广播fanout，定向direct，通配符topic）
a. 发布与订阅模式：使用了fanout类型的交换机，可以将一个消息发送到所有与交换机绑定的队列并被消费者接收。
1package cn.lemon.rabbitmq.ps; 2 3 4import cn.lemon.rabbitmq.utils.ConnectionUtil; 5import com.rabbitmq.client.BuiltinExchangeType; 6import com.rabbitmq.client.Channel; 7import com.rabbitmq.client.Connection; 8 9 10/* 11 消息生产者：发布订阅模式 12 发布订阅模式Publish/subscribe 13 1.需要设置类型为fanout的交换机 14 2.并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列 15 */ 16public class Producer { 17 18 // 交换机名称 19 public static final String FANOUT_EXCHANGE = "fanout_exchange"; 20 // 队列名称1 21 public static final String FANOUT_QUEUE_1 = "fanout_queue_1"; 22 // 队列名称2 23 public static final String FANOUT_QUEUE_2 = "fanout_queue_2"; 24 25 public static void main(String[] args) throws Exception { 26 // 1.创建连接 27 Connection connection = ConnectionUtil.getConnection(); 28 // 2.创建频道 29 Channel channel = connection.createChannel(); 30 31 /** 32 * 3.声明交换机 33 * 参数1：交换机名称 34 * 参数2：交换机类型：fanout、direct、topic、headers 35 */ 36 channel.exchangeDeclare(FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT); 37 38 /** 39 * 4.声明（创建）队列 40 * 参数1：队列名称 41 * 参数2：是否定义持久化队列 42 * 参数3：是否独占本次连接 43 * 参数4：是否在不使用的时候自动删除队列 44 * 参数5：队列其它参数 45 */ 46 channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null); 47 channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null); 48 49 // 5.队列绑定交换机 50 channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHANGE, ""); 51 channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHANGE, ""); 52 53 // 6. 发送多个消息 54 for (int i = 1; i <= 10; i++) { 55 // 要发送的信息 56 String message = "你好；小兔子！" + i; 57 /** 58 * 参数1：交换机名称，如果没有指定则使用默认Default Exchage 59 * 参数2：路由key,简单模式可以传递队列名称 60 * 参数3：消息其它属性 61 * 参数4：消息内容 62 */ 63 channel.basicPublish(FANOUT_EXCHANGE, "", null, message.getBytes()); 64 System.out.println("已发送消息：" + message); 65 } 66 // 7. 关闭资源 67 channel.close(); 68 connection.close(); 69 70 } 71 72 73} 74 75package cn.lemon.rabbitmq.ps; 76 77import cn.lemon.rabbitmq.utils.ConnectionUtil; 78import cn.lemon.rabbitmq.utils.ConsumerUtil; 79import com.rabbitmq.client.Channel; 80import com.rabbitmq.client.Connection; 81import com.rabbitmq.client.DefaultConsumer; 82 83/** 84 * 消息的消费者，通过设置监听自动获取队列中的消息，实现消费 85 */ 86public class Consumer1 { 87 88 public static void main(String[] args) throws Exception { 89 // 1.创建连接 90 Connection connection = ConnectionUtil.getConnection(); 91 92 // 2.创建频道 93 Channel channel = connection.createChannel(); 94 95 // 3.申明(创建)交换机 96 // 队列绑定到交换机，只要在生产者绑定，消费者可以不用再绑定 97 //channel.exchangeDeclare(Producer.FANOUT_EXCHANGE,BuiltinExchangeType.FANOUT); 98 99 // 4.声明（创建）队列 100 //channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null); 101 102 // 5.队列绑定到交换机 103 //channel.queueBind(Producer.FANOUT_QUEUE_1,Producer.FANOUT_EXCHANGE,""); 104 105 // 6.创建消费者；并设置消息处理 106 DefaultConsumer consumer = ConsumerUtil.getConsumer(channel); 107 108 // 7.监听消息 109 channel.basicConsume(Producer.FANOUT_QUEUE_1, true, consumer); 110 } 111} 112 113package cn.lemon.rabbitmq.ps; 114 115import cn.lemon.rabbitmq.utils.ConnectionUtil; 116import cn.lemon.rabbitmq.utils.ConsumerUtil; 117import com.rabbitmq.client.BuiltinExchangeType; 118import com.rabbitmq.client.Channel; 119import com.rabbitmq.client.Connection; 120import com.rabbitmq.client.DefaultConsumer; 121 122/** 123 * 消息的消费者，通过设置监听自动获取队列中的消息，实现消费 124 */ 125public class Consumer2 { 126 127 public static void main(String[] args) throws Exception { 128 // 1.创建连接 129 Connection connection = ConnectionUtil.getConnection(); 130 131 // 2.创建频道 132 Channel channel = connection.createChannel(); 133 134 // 3.创建交换机 135 channel.exchangeDeclare(Producer.FANOUT_EXCHANGE,BuiltinExchangeType.FANOUT); 136 137 // 4.声明（创建）队列 138 channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null); 139 140 // 5.队列绑定到交换机 141 channel.queueBind(Producer.FANOUT_QUEUE_2,Producer.FANOUT_EXCHANGE,""); 142 143 // 6.创建消费者；并设置消息处理 144 DefaultConsumer consumer = ConsumerUtil.getConsumer(channel); 145 146 // 7.监听消息 147 channel.basicConsume(Producer.FANOUT_QUEUE_2, true, consumer); 148 } 149}
b. 路由模式：使用了direct类型的交换机，可以将一个消息发送到routing key相关的队列并被消费者接收。
1package cn.lemon.rabbitmq.routing; 2 3import cn.lemon.rabbitmq.utils.ConnectionUtil; 4import com.rabbitmq.client.BuiltinExchangeType; 5import com.rabbitmq.client.Channel; 6import com.rabbitmq.client.Connection; 7 8/* 9 路由消息生产者：消息发送到交换机 10 生产者发送两个消息（路由key分别为：insert、update） 11 创建两个消费者，分别绑定的队列中路由为（insert，update） 12 */ 13public class Producer { 14 15 //交换机名 16 public static final String DIRECT_EXCHANGE = "direct_exchange"; 17 //队列名 18 public static final String DIRECT_QUEUE_INSERT = "direct_queue_insert"; 19 public static final String DIRECT_QUEUE_UPDATE = "direct_queue_update"; 20 21 public static void main(String[] args) throws Exception { 22 //1. 创建连接 23 Connection connection = ConnectionUtil.getConnection(); 24 //2. 创建频道 25 Channel channel = connection.createChannel(); 26 /** 27 * 3.声明交换机 28 * 参数1：交换机名称 29 * 参数2：交换机类型：fanout、direct、topic、headers 30 */ 31 channel.exchangeDeclare(DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT); 32 /** 33 * 4.声明（创建）队列 34 * 参数1：队列名称 35 * 参数2：是否定义持久化队列 36 * 参数3：是否独占本次连接 37 * 参数4：是否在不使用的时候自动删除队列 38 * 参数5：队列其它参数 39 */ 40 channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null); 41 channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null); 42 43 /** 44 * 5.队列绑定交换机 45 * 参数1：队列名 46 * 参数2：交换机名 47 * 参数3：路由key 48 */ 49 channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHANGE, "insert"); 50 channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHANGE, "update"); 51 52 // 6.发送消息 53 String message = "新增了商品，路由模式；routing key 为 insert "; 54 channel.basicPublish(DIRECT_EXCHANGE, "insert", null, message.getBytes()); 55 System.out.println("已发送消息：" + message); 56 57 message = "修改了商品，路由模式；routing key 为 update "; 58 channel.basicPublish(DIRECT_EXCHANGE, "update", null, message.getBytes()); 59 System.out.println("已发送消息：" + message); 60 61 // 7.关闭资源 62 channel.close(); 63 connection.close(); 64 } 65 66} 67 68package cn.lemon.rabbitmq.routing; 69 70import cn.lemon.rabbitmq.utils.ConnectionUtil; 71import cn.lemon.rabbitmq.utils.ConsumerUtil; 72import com.rabbitmq.client.Channel; 73import com.rabbitmq.client.Connection; 74import com.rabbitmq.client.DefaultConsumer; 75 76/** 77 * 消息消费者: 消费队列中的消息 78 * 消息路由为insert 79 */ 80public class Consumer1 { 81 82 public static void main(String[] args) throws Exception { 83 // 1.创建连接 84 Connection connection = ConnectionUtil.getConnection(); 85 86 // 2.创建频道 87 Channel channel = connection.createChannel(); 88 89 // 3.创建交换机 生产者已经创建了，可以不用在创建 90 //channel.exchangeDeclare(Producer.DIRECT_EXCHANGE,BuiltinExchangeType.DIRECT); 91 92 // 4.声明（创建）队列 93 channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null); 94 95 // 5.队列绑定到交换机 96 channel.queueBind(Producer.DIRECT_QUEUE_INSERT, Producer.DIRECT_EXCHANGE, "insert"); 97 98 // 6.创建消费者,并设置消息处理 99 DefaultConsumer consumer = ConsumerUtil.getConsumer(channel); 100 101 // 7.监听消息 102 channel.basicConsume(Producer.DIRECT_QUEUE_INSERT, true, consumer); 103 } 104} 105 106package cn.lemon.rabbitmq.routing; 107 108import cn.lemon.rabbitmq.utils.ConnectionUtil; 109import cn.lemon.rabbitmq.utils.ConsumerUtil; 110import com.rabbitmq.client.BuiltinExchangeType; 111import com.rabbitmq.client.Channel; 112import com.rabbitmq.client.Connection; 113import com.rabbitmq.client.DefaultConsumer; 114 115/** 116 * 消息消费者: 消费队列中的消息 117 * 消息路由为update 118 */ 119public class Consumer2 { 120 121 public static void main(String[] args) throws Exception { 122 // 1.创建连接 123 Connection connection = ConnectionUtil.getConnection(); 124 125 // 2.创建频道 126 Channel channel = connection.createChannel(); 127 128 // 3.创建交换机 129 channel.exchangeDeclare(Producer.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT); 130 131 // 4.声明（创建）队列 132 channel.queueDeclare(Producer.DIRECT_QUEUE_UPDATE, true, false, false, null); 133 134 // 5.队列绑定到交换机 135 channel.queueBind(Producer.DIRECT_QUEUE_UPDATE, Producer.DIRECT_EXCHANGE, "update"); 136 137 // 6.创建消费者；并设置消息处理 138 DefaultConsumer consumer = ConsumerUtil.getConsumer(channel); 139 140 // 7.监听消息 141 channel.basicConsume(Producer.DIRECT_QUEUE_UPDATE, true, consumer); 142 } 143}
c. 通配符模式：使用了topic类型的交换机，可以将一个消息发送到routing key（*,#）相关的队列并被消费者接收。
1package cn.lemon.rabbitmq.topic; 2 3 4import cn.lemon.rabbitmq.utils.ConnectionUtil; 5import com.rabbitmq.client.BuiltinExchangeType; 6import com.rabbitmq.client.Channel; 7import com.rabbitmq.client.Connection; 8 9/** 10 * Topic通配符模型消息生产者：消息发送到交换机 11 * 生产者：发送包含有item.insert，item.update，item.delete的3种路由key的消息 12 * RoutingKey 一般都是有一个或多个单词组成，多个单词之间以”.”分割 13 * 通配符规则： 14 * # 匹配一个或多个词 15 * * 匹配一个词 16 */ 17public class Producer { 18 19 // 交换机名称 20 public static final String TOPIC_EXCHANGE = "topic_exchange"; 21 // 队列名称 22 public static final String TOPIC_QUEUE_1 = "topic_queue_1"; 23 // 队列名称 24 public static final String TOPIC_QUEUE_2 = "topic_queue_2"; 25 26 public static void main(String[] args) throws Exception { 27 //1. 创建连接 28 Connection connection = ConnectionUtil.getConnection(); 29 30 //2. 创建频道 31 Channel channel = connection.createChannel(); 32 33 //3. 声明交换机 参数1(交换机名) 参数2(交换机类型) 34 channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC); 35 36 //4. 声明队列 消费者已经声明，生产者可以不用声明 37 //channel.queueDeclare(TOPIC_QUEUE_1, true, false, false, null); 38 39 // 5.队列绑定到交换机 消费者已经绑定，生产者不用绑定 40 //channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHANGE,"item.*"); 41 42 //5.发送消息 43 String message = "新增了商品，Topic模式，路由key为item.insert"; 44 channel.basicPublish(TOPIC_EXCHANGE, "item.insert", null, message.getBytes()); 45 System.out.println("已发送消息：" + message); 46 47 message = "修改了商品，Topic模式，路由key为item.update"; 48 channel.basicPublish(TOPIC_EXCHANGE, "item.update", null, message.getBytes()); 49 System.out.println("已发送消息：" + message); 50 51 message = "删除了商品，Topic模式，路由key为item.delete"; 52 channel.basicPublish(TOPIC_EXCHANGE, "item.delete", null, message.getBytes()); 53 System.out.println("已发送消息：" + message); 54 55 // 6.关闭资源 56 channel.close(); 57 connection.close(); 58 } 59 60} 61 62package cn.lemon.rabbitmq.topic; 63 64import cn.lemon.rabbitmq.utils.ConnectionUtil; 65import cn.lemon.rabbitmq.utils.ConsumerUtil; 66import com.rabbitmq.client.Channel; 67import com.rabbitmq.client.Connection; 68import com.rabbitmq.client.DefaultConsumer; 69 70public class Consumer1 { 71 72 public static void main(String[] args) throws Exception { 73 //1. 创建连接 74 Connection connection = ConnectionUtil.getConnection(); 75 76 //2. 创建频道 77 Channel channel = connection.createChannel(); 78 79 //3. 声明交换机 参数1(交换机名) 参数2(交换机类型) 80 //channel.exchangeDeclare(Producer.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC); 81 82 //4. 声明队列 生产者已经创建了，消费者可以不用创建 83 channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null); 84 85 // 5.队列绑定到交换机 86 channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHANGE, "item.*"); 87 88 // 6.创建消费者；并设置消息处理 89 DefaultConsumer consumer = ConsumerUtil.getConsumer(channel); 90 91 // 7.监听消息 92 channel.basicConsume(Producer.TOPIC_QUEUE_1, true, consumer); 93 94 } 95} 96 97package cn.lemon.rabbitmq.topic; 98 99import cn.lemon.rabbitmq.utils.ConnectionUtil; 100import cn.lemon.rabbitmq.utils.ConsumerUtil; 101import com.rabbitmq.client.Channel; 102import com.rabbitmq.client.Connection; 103import com.rabbitmq.client.DefaultConsumer; 104 105/** 106 * 消息消费者: 消费队列中的消息 107 */ 108public class Consumer2 { 109 110 public static void main(String[] args) throws Exception { 111 // 1.创建连接 112 Connection connection = ConnectionUtil.getConnection(); 113 114 // 2.创建频道 115 Channel channel = connection.createChannel(); 116 117 // 3.创建交换机 118 //channel.exchangeDeclare(Producer.TOPIC_EXCHANGE,BuiltinExchangeType.TOPIC); 119 120 // 4.声明（创建）队列 消费者创建，生产者没有创建 121 channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null); 122 123 // 5.队列绑定到交换机 124 channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHANGE, "item.update"); 125 channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHANGE, "item.delete"); 126 127 // 6.创建消费者；并设置消息处理 128 DefaultConsumer consumer = ConsumerUtil.getConsumer(channel); 129 130 // 7.监听消息 131 channel.basicConsume(Producer.TOPIC_QUEUE_2, true, consumer); 132 } 133}