目录
引言
实现 VirtualHost 类
属性
交换机相关操作
队列相关操作
绑定相关操作
消息相关操作
关于线程安全问题
针对 VirtualHost 单元测试
引言
- 虚拟主机的概念类似于 MySQL 的 database，用于将 交换机、队列、绑定、消息 进行逻辑上的隔离
- 虚拟主机不仅仅要管理数据，还需要提供一些 核心 API，供上层代码进行调用
- 就是将之前写的 内存 和 硬盘 的数据管理给串起来
- 即整个程序的核心业务逻辑
核心 API：
- 创建交换机 exchangeDeclare
- 删除交换机 exchangeDelete
- 创建队列 queueDeclare
- 删除队列 queueDelete
- 创建绑定 queueBind
- 删除绑定 queueUnbind
- 发送消息 basicPublish
- 订阅消息 basicConsume
- 确认消息 basicAck
注意点一：
- 此处各 API 的取名，各 API 中设定的参数，均参考于RabbitMQ
注意点二：
- 此处我们仅实现单个虚拟主机，并不打算实现添加/删除虚拟主机的 API
- 但是会在设计数据结构上留下这样的扩展空间
实例理解
- 虚拟主机存在目的，就是为了保证隔离，即不同虚拟主机之间的内容互不影响
- 当 虚拟主机1中创建了一个名为 “testExchange” 的交换机
- 而 虚拟主机2 中也创建了一个名为 “testExchange” 的交换机
- 虽然这两个交换机的名字相同，但是却处于不同虚拟主机中，所以需要区分开来
问题：
- 如何表示 交换机 与 虚拟主机 之间的从属关系？
可选方案：
- 方案一：参考数据库设计 “一对多” 的方案，给交换机表添加个属性，虚拟主机 id/name
- 方案二：重新约定交换机的名字，即 新交换机名字 = 虚拟主机名字 + 交换机真实名字
- 方案三：给每个虚拟主机，分配一组不同的数据库和文件（比方案二麻烦，但更优雅）
回答：
- 此处我们选择方案二！
- 约定在 VirtualHost 中的核心 api 里，对exchangeName 和 queueName 做一个转换
- 在该层代码中进行转换后，后续代码 MemoryDataCenter、DiskDataCenter 无需调整
- 按照这个方式，也可以去区分不同的队列
- 绑定 与 交换机和队列 相关，通过上述操作，绑定自然也就被隔离开了！
- 消息 与 队列 相关，因为队列名已经区分开了，消息自然也就被区分开了！
实现 VirtualHost 类
属性
@Getter public class VirtualHost { private String virtualHostName; private MemoryDataCenter memoryDataCenter = new MemoryDataCenter(); private DiskDataCenter diskDataCenter = new DiskDataCenter(); private Router router = new Router(); private ConsumerManager consumerManager = new ConsumerManager(this); // 操作交换机的锁对象 private final Object exchangeLocker = new Object(); // 操作队列的锁对象 private final Object queueLocker = new Object(); public VirtualHost(String name) { this.virtualHostName = name; // 对于 MemoryDataCenter 来说，不需要额外的初始化操作的，只要对象 new 出来就行 // 但是，针对 DiskDataCenter 来说，则需要进行初始化操作，建库建表和初始化数据的设定 // 另外还需要针对硬盘的数据，进行恢复到内存中 diskDataCenter.init(); try { memoryDataCenter.recovery(diskDataCenter); } catch (IOException | MqException | ClassNotFoundException e) { e.printStackTrace(); System.out.println("[VirtualHost] 恢复内存数据失败！"); } } }
- Router 类用于实现交换机的转发规则，验证 bindingKey 和 routingKey 的合法性
- ConsumerManager 类用于实现消费消息的核心逻辑
交换机相关操作
// 创建交换机 // 如果交换机不存在就创建，如果存在 则直接返回 // 返回值是 boolean，创建成功，返回 true，失败返回 false public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map arguments) { // 把交换机的名字，加上虚拟主机作为前缀 exchangeName = virtualHostName + exchangeName; try { synchronized (exchangeLocker) { // 1、判定该交换机是否已经存在，直接通过内存查询 Exchange existsExchange = memoryDataCenter.getExchange(exchangeName); if (existsExchange != null) { // 该交换机已经存在 System.out.println("[VirtualHost] 交换机已经存在！exchangeName = " + exchangeName); return true; } // 2、真正创建交换机，先构造 Exchange 对象 Exchange exchange = new Exchange(); exchange.setName(exchangeName); exchange.setType(exchangeType); exchange.setDurable(durable); exchange.setAutoDelete(autoDelete); exchange.setArguments(arguments); // 3、把交换机对象写入硬盘 if(durable) { diskDataCenter.insertExchange(exchange); } // 4、把交换机对象写入内存 memoryDataCenter.insertExchange(exchange); System.out.println("[VirtualHost] 交换机创建完成！exchangeName = " + exchangeName); // 上述逻辑，先写硬盘，后写内存，目的就是因为硬盘更容易写失败，如果硬盘写失败了，内存就不写了 // 要是先写内存，内存写成功了，硬盘写失败了，还需要把内存的数据给再删掉，就比较麻烦了 } return true; }catch (Exception e) { System.out.println("[VirtualHost] 交换机创建失败！exchangeName = " + exchangeName); e.printStackTrace(); return false; } } // 删除交换机 public boolean exchangeDelete(String exchangeName) { exchangeName = virtualHostName + exchangeName; try { synchronized (exchangeLocker) { // 1、先找到对应的交换机 Exchange toDelete = memoryDataCenter.getExchange(exchangeName); if(toDelete == null) { throw new MqException("[VirtualHost] 交换机不存在无法删除！"); } // 2、删除硬盘上的数据 if(toDelete.isDurable()) { diskDataCenter.deleteExchange(exchangeName); } // 3、删除内存中的交换机数据 memoryDataCenter.deleteExchange(exchangeName); System.out.println("[VirtualHost] 交换机删除成功！exchangeName = " + exchangeName); } return true; }catch (Exception e) { System.out.println("[VirtualHost] 交换机删除失败！exchangeName = " + exchangeName); e.printStackTrace(); return false; } }
队列相关操作
// 创建队列 public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete, Map arguments) { // 把队列的名字，给拼接上虚拟主机的名字 queueName = virtualHostName + queueName; try { synchronized (queueLocker) { // 1、判定队列是否存在 MSGQueue existsQueue = memoryDataCenter.getQueue(queueName); if(existsQueue != null) { System.out.println("[VirtualHost] 队列已经存在！queueName = " + queueName); return true; } // 2、创建队列对象 MSGQueue queue = new MSGQueue(); queue.setName(queueName); queue.setDurable(durable); queue.setExclusive(exclusive); queue.setAutoDelete(autoDelete); queue.setArguments(arguments); // 3、写硬盘 if(durable) { diskDataCenter.insertQueue(queue); } // 4、写内存 memoryDataCenter.insertQueue(queue); System.out.println("[VirtualHost] 队列创建成功！queueName = " + queueName); } return true; }catch (Exception e) { System.out.println("[VirtualHost] 队列创建失败！queueName = " + queueName); e.printStackTrace(); return false; } } // 删除队列 public boolean queueDelete(String queueName) { queueName = virtualHostName + queueName; try { synchronized (queueLocker) { // 1、根据队列名字，查询下当前的队列对象 MSGQueue queue = memoryDataCenter.getQueue(queueName); if(queue == null) { throw new MqException("[VirtualHost] 队列不存在！无法删除！queueName = " + queueName); } // 2、删除硬盘数据 if(queue.isDurable()) { diskDataCenter.deleteQueue(queueName); } // 3、删除内存数据 memoryDataCenter.deleteQueue(queueName); System.out.println("[VirtualHost] 删除队列成功！queueName = " + queueName); } return true; }catch (Exception e) { System.out.println("[VirtualHost] 删除队列失败！queueName = " + queueName); e.printStackTrace(); return false; } }
绑定相关操作
public boolean queueBind(String queueName, String exchangeName, String bindingKey) { queueName = virtualHostName + queueName; exchangeName = virtualHostName + exchangeName; try { synchronized (exchangeLocker) { synchronized (queueLocker) { // 1、判定当前的绑定是否已经存在了 Binding exchangeBinding = memoryDataCenter.getBinding(exchangeName,queueName); if(exchangeBinding != null) { throw new MqException("[VirtualHost] binding 已经存在！ queueName = " + queueName + ", exchangeName = " + exchangeName); } // 2、验证 bindingKey 是否合法 if(!router.checkBindingKey(bindingKey)) { throw new MqException("[VirtualHost] bindingKey 非法！ bindingKey = " + bindingKey); } // 3、创建 Binding 对象 Binding binding = new Binding(); binding.setExchangeName(exchangeName); binding.setQueueName(queueName); binding.setBindingKey(bindingKey); // 4、获取一下对应的交换机和队列，如果交换机或者队列不存在，这样的绑定也是无法创建的 MSGQueue queue = memoryDataCenter.getQueue(queueName); if(queue == null) { throw new MqException("[VirtualHost] 队列不存在！ queueName = " + queueName); } Exchange exchange = memoryDataCenter.getExchange(exchangeName); if(exchange == null) { throw new MqException("[VirtualHost] 交换机不存在！ exchangeName = " + exchangeName); } // 5、先写硬盘 if(queue.isDurable() && exchange.isDurable()) { diskDataCenter.insertBinding(binding); } // 6、写入内存 memoryDataCenter.insertBinding(binding); System.out.println("[VirtualHost] 绑定创建成功！queueName = " + queueName + ", exchangeName = " + exchangeName); } } return true; }catch (Exception e) { System.out.println("[VirtualHost] queueBind 失败！queueName = " + queueName + ", exchangeName = " + exchangeName); e.printStackTrace(); return false; } } public boolean queueUnbind(String queueName, String exchangeName) { queueName = virtualHostName + queueName; exchangeName = virtualHostName + exchangeName; try { synchronized (exchangeLocker) { synchronized (queueLocker) { // 1、获取 binding 看是否已经存在 Binding binding = memoryDataCenter.getBinding(exchangeName,queueName); if(binding == null) { throw new MqException("[VirtualHost] 删除绑定失败！绑定不存在！queueName = " + queueName + ", exchangeName = " + exchangeName); } // 2、无论绑定是否持久化了，都尝试从硬盘删一下，就算不存在，这个删除也无副作用 diskDataCenter.deleteBinding(binding); // 3、删除内存的数据 memoryDataCenter.deleteBinding(binding); System.out.println("[VirtualHost] 删除绑定成功！"); } } return true; }catch (Exception e) { System.out.println("[VirtualHost] 删除绑定失败！！"); e.printStackTrace(); return false; } }注意：
- 观察下图红框代码
问题：
- 如果在解除该绑定之前，该绑定中的 交换机 或 队列 已经被删除了
- 那么此时我们采用上述逻辑就无法解除绑定了
方案一：
- 参考类似于 MySQL的外键一样，删除队列/交换机的时候，判定一下看当前交换机/队列 是否存在对应的绑定，如果存在，则禁止删除队列/交换机，要求先解除绑定，再尝试删除队列/交换机
- 优点：更严谨
- 缺点：更麻烦，尤其是，查看当前的队列是否有对应的绑定的时候
- 由上图我们给 Binding设定的内存数据结构可知，查看一个交换机有哪些绑定是比较容易的，但是查看一个队列有哪些绑定是比较难的！
方案二：
- 删除绑定时，干脆不校验交换机/队列存在，直接就尝试删除
- 优点：简单
- 缺点：没那么严谨，也还好
回答：
- 此处，我们直接采取第二种方法！
- 无论 exchange 或 queue 是否持久化了，均尝试从硬盘上删除一下
- 因为即使 exchange 或 queue 未持久化到硬盘上，底层调用的 delete 语句也不会有什么副作用
消息相关操作
// 发送消息到指定的交换机/队列中 public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body) { try { // 1、转换交换机的名字 exchangeName = virtualHostName + exchangeName; // 2、检查 routingKey 是否合法 if(!router.checkRoutingKey(routingKey)) { throw new MqException("[VirtualHost] routingKey 非法！routingKey = " + rou服务器托管网tingKey); } // 3、查找交换机对象 Exchange exchange = memoryDataCenter.getExchange(exchangeName); if(exchange == null) { throw new MqException("[VirtualHost] 交换机不存在！exchangeName = " + exchangeName); } // 4、判定交换机类型 if(exchange.getType() == ExchangeType.DIRECT) { // 按照直接交换机的方式来转发消息 // 以 routingKey 作为队列的名字，直接把消息写入到指定的队列中 // 此时，可以无视绑定关系 String queueName = virtualHostName + routingKey; // 5、构造消息对象 Message message = Message.createMessageWithId(routingKey,basicProperties,body); // 6、查找该队列对应的对象 MSGQueue queue = memoryDataCenter.getQueue(queueName); if(queue == null) { throw new MqException("[VirtualHost] 队列不存在！queueName = " + queueName); } // 7、队列存在，直接给队列中写入消息 sendMessage(queue,message); }else { // 按照 fanout 和 topic 的方式来转发 // 5、找到该交换机关联的所有绑定，并遍历这些绑定对象 ConcurrentHashMap bindingsMap = memoryDataCenter.getBindings(exchangeName); for (Map.Entry entry : bindingsMap.entrySet()) { // 1）获取到绑定对象，判定对应的的队列是否存在 Binding binding = entry.getValue(); MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName()); if(queue == null) { // 此处无需抛出异常，可能此处有多个这样的队列 // 希望不要因为一个队列影响到其他队列的消息的传输 System.out.print服务器托管网ln("[VirtualHost] basicPublish 发送消息时，队列不存在！queueName = " + binding.getQueueName()); continue; } // 2）构造消息对象 Message message = Message.createMessageWithId(routingKey,basicProperties,body); // 3）判定这个消息时否能转发给该队列 // 如果是 fanout，所有绑定的队列都要转发的 // 如果是 topic，还需要判定下，bindingKey 和 routingKey 是不是匹配 if(!router.route(exchange.getType(),binding,message)){ continue; } // 真正转发消息给队列 sendMessage(queue,message); } } return true; }catch (Exception e) { System.out.println("[VirtualHost] 消息发送失败！"); e.printStackTrace(); return false; } } private void sendMessage(MSGQueue queue,Message message) throws IOException, MqException, InterruptedException { // 此处发送消息，就是把消息写入到硬盘 和 内存上 int deliverMode = message.getDeliverMode(); // deliverMode 为 1 表示不持久化，deliverMode 为 2 表示持久化 if(deliverMode == 2) { diskDataCenter.sendMessage(queue,message); } // 写入内存 memoryDataCenter.sendMessage(queue,message); // 通知消费者可以消费消息了 consumerManager.notifyConsume(queue.getName()); } // 订阅消息 // 添加一个队列的订阅者，当队列收到消息之后，就要把消息推送给对应的订阅者 // consumerTag：表示消费者的身份标识 // autoAck：消息被消费完成后，应答的方式，为 true 自动应答，为 false 手动应答 // consumer：是一个回调函数，此处类型设定成函数式接口，这样后续调用 basicConsume 并且传实参的时候，就可以写作 lambda 样子 public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) { // 构造一个 ConsumerEnv 对象，把这个对应的队列找到，再把这个 Consumer 对象添加到该队列中 queueName = virtualHostName + queueName; try { consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer); System.out.println("[VirtualHost] basicConsume 成功！ queueName = " + queueName); return true; }catch (Exception e) { System.out.println("[VirtualHost] basicConsume 失败！ queueName = " + queueName); e.printStackTrace(); return false; } } public boolean basicAck(String queueName,String messageId) { queueName = virtualHostName + queueName; try { // 1、获取到消息和队列 Message message = memoryDataCenter.getMessage(messageId); if(message == null) { throw new MqException("[VirtualHost] 要确认的消息不存在！messageId = " + messageId); } MSGQueue queue = memoryDataCenter.getQueue(queueName); if(queue == null) { throw new MqException("[VirtualHost] 要确认的消息不存在！queueName = " + queueName); } // 2、删除硬盘上的数据 if(message.getDeliverMode() == 2) { diskDataCenter.deleteMessage(queue,message); } // 3、删除消息中心的数据 memoryDataCenter.removeMessage(messageId); // 4、删除待确认的集合中的消息 memoryDataCenter.removeMessageWaitAck(queueName,messageId); System.out.println("[VirtualHost] basicAck 成功！消息被确认成功！queueName = " + queueName + ", messageId = " + messageId); return true; }catch (Exception e) { System.out.println("[VirtualHost] basicAck 失败！消息确认失败！queueName = " + queueName + ", messageId = " + messageId); e.printStackTrace(); return false; } }
关于线程安全问题
- 针对创建交换机 exchangeDeclare 方法加锁
- 针对删除交换机exchangeDelete 方法加锁
- 如上图所示，此处我们是针对 exchangeLocker 对象进行加锁的，从而导致这个锁的粒度还是比较大的
- 比如 创建/删除 交换机A 时，此时就会影响到交换机B 的创建/删除
注意：
- 此处我们确实可以做出一系列调整，加一个更细粒度的锁，但是也没啥必要
- 对于 Broker Server 来说，创建交换机、创建绑定、创建队列、删除交换机、删除绑定、删除队列，这些均属于低频操作！
- 既然是低频操作，所以遇到两个线程都去操作创建队列之类的情况本身就概率很低了
- 因此，对于绝大多数情况来说，是不会触发锁冲突的
- 再加之 synchronized 最初为偏向锁状态，该状态下加锁成本也还好，只有遇到竞争才会真正加锁
- 当然，为了应对一些少数的极端情况，此处加锁还是有一定必要的
问题：
- 既然在这一层代码加锁了
- 里面的 MemoryDataCenter 中的操作是否就不必加锁了？
- 是否之前的加锁就没有意义了？
回答：
- 我们并不知道MemoryDataCenter 的方法是给哪个类进行调用的
- 因为当前VirtualHost 自身是保证了线程安全的
- 所以VirtualHost 内部调用的 MemoryDataCenter 中不加锁也问题不大
- 但是如果是另一个自身未保证线程安全的类，也多线程调用 MemoryCenter 呢？
针对 VirtualHost 单元测试
- 编写测试用例代码是十分重要的！
package com.example.demo; import com.example.demo.common.Consumer; import com.example.demo.mqserver.VirtualHost; import com.example.demo.mqserver.core.BasicProperties; import com.example.demo.mqserver.core.ExchangeType; import org.apache.tomcat.util.http.fileupload.FileUtils; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; import org.springframework.boot.test.context.SpringBootTest; import java.io.File; import java.io.IOException; @SpringBootTest public class VirtualHostTests { private VirtualHost virtualHost = null; @BeforeEach public void setUp() { DemoApplication.context = SpringApplication.run(DemoApplication.class); virtualHost = new VirtualHost("default"); } @AfterEach public void tearDown() throws IOException { DemoApplication.context.close(); virtualHost = null; // 把硬盘的目录删除掉 File dataDir = new File("./data"); FileUtils.deleteDirectory(dataDir); } @Test public void testExchangeDeclare() { boolean ok = virtualHost.exchangeDeclare("testExchange", ExchangeType.DIRECT, true,false,null); Assertions.assertTrue(ok); } @Test public void testExchangeDelete() { boolean ok = virtualHost.exchangeDeclare("testExchange", ExchangeType.DIRECT, true,false,null); Assertions.assertTrue(ok); ok = virtualHost.exchangeDelete("testExchange"); Assertions.assertTrue(ok); } @Test public void testQueueDeclare() { boolean ok = virtualHost.queueDeclare("testQueue",true, false,false,null); Assertions.assertTrue(ok); } @Test public void testQueueDelete() { boolean ok = virtualHost.queueDeclare("testQueue",true, false,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueDelete("testQueue"); Assertions.assertTrue(ok); } @Test public void testQueueBind() { boolean ok = virtualHost.exchangeDeclare("testExchange", ExchangeType.DIRECT, true,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueDeclare("testQueue",true, false,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueBind("testQueue","testExchange","testBindingKey"); Assertions.assertTrue(ok); } @Test public void testQueueUnbind() { boolean ok = virtualHost.exchangeDeclare("testExchange", ExchangeType.DIRECT, true,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueDeclare("testQueue",true, false,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueBind("testQueue","testExchange","testBindingKey"); Assertions.assertTrue(ok); ok = virtualHost.queueUnbind("testQueue","testExchange"); Assertions.assertTrue(ok); } @Test public void testBasicPublish() { boolean ok = virtualHost.exchangeDeclare("testExchange", ExchangeType.DIRECT, true,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueDeclare("testQueue",true, false,false,null); Assertions.assertTrue(ok); ok = virtualHost.basicPublish("testExchange","testQueue",null, "hello".getBytes()); Assertions.assertTrue(ok); } // 先订阅队列，后发送消息 @Test public void testBasicConsume1() throws InterruptedException { boolean ok = virtualHost.exchangeDeclare("testExchange", ExchangeType.DIRECT, true,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueDeclare("testQueue",true, false,false,null); Assertions.assertTrue(ok); // 先订阅队列 ok = virtualHost.basicConsume("testConsumerTag", "testQueue", true, new Consumer() { @Override public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) { // 消费者自身设定的回调方法 System.out.println("messageId = " + basicProperties.getMessageId()); System.out.println("body = " + new String(body,0,body.length)); Assertions.assertEquals("testQueue",basicProperties.getRoutingKey()); Assertions.assertEquals(1,basicProperties.getDeliverMode()); Assertions.assertArrayEquals("hello".getBytes(),body); } }); Assertions.assertTrue(ok); Thread.sleep(500); // 再发送消息 ok = virtualHost.basicPublish("testExchange","testQueue",null, "hello".getBytes()); Assertions.assertTrue(ok); } // 先发送消息，后订阅队列 @Test public void testBasicConsume2() throws InterruptedException { boolean ok = virtualHost.exchangeDeclare("testExchange", ExchangeType.DIRECT, true,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueDeclare("testQueue",true, false,false,null); Assertions.assertTrue(ok); // 先发送消息 ok = virtualHost.basicPublish("testExchange","testQueue",null, "hello".getBytes()); Assertions.assertTrue(ok); // 再订阅队列 ok = virtualHost.basicConsume("testConsumerTag", "testQueue", true, new Consumer() { @Override public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) { // 消费者自身设定的回调方法 System.out.println("messageId = " + basicProperties.getMessageId()); System.out.println("body = " + new String(body,0,body.length)); Assertions.assertEquals("testQueue",basicProperties.getRoutingKey()); Assertions.assertEquals(1,basicProperties.getDeliverMode()); Assertions.assertArrayEquals("hello".getBytes(),body); } }); Assertions.assertTrue(ok); Thread.sleep(500); } @Test public void testBasicConsumeFanout() throws InterruptedException { boolean ok = virtualHost.exchangeDeclare("testExchange", ExchangeType.FANOUT, false,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueDeclare("testQueue1",false, false,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueBind("testQueue1","testExchange",""); Assertions.assertTrue(ok); ok = virtualHost.queueDeclare("testQueue2",false, false,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueBind("testQueue2","testExchange",""); Assertions.assertTrue(ok); // 往交换机中发送一个消息 ok = virtualHost.basicPublish("testExchange","",null,"hello".getBytes()); Assertions.assertTrue(ok); Thread.sleep(500); // 两个消费者订阅上述的两个队列 ok = virtualHost.basicConsume("testConsumer", "testQueue1", true, new Consumer() { @Override public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) { System.out.println("consumerTag = " + consumerTag); System.out.println("messageId = " + basicProperties.getMessageId()); Assertions.assertArrayEquals("hello".getBytes(),body); } }); Assertions.assertTrue(ok); ok = virtualHost.basicConsume("testConsumer2", "testQueue2", true, new Consumer() { @Override public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) { System.out.println("consumerTag = " + consumerTag); System.out.println("messageId = " + basicProperties.getMessageId()); Assertions.assertArrayEquals("hello".getBytes(),body); } }); Assertions.assertTrue(ok); Thread.sleep(500); } @Test public void testBasicConsumeTopic() throws InterruptedException { boolean ok = virtualHost.exchangeDeclare("testExchange", ExchangeType.TOPIC, false,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueDeclare("testQueue1",false, false,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueBind("testQueue1","testExchange","aaa.*.bbb"); Assertions.assertTrue(ok); ok = virtualHost.basicPublish("testExchange","aaa.ccc.bbb",null,"hello".getBytes()); Assertions.assertTrue(ok); ok = virtualHost.basicConsume("testConsumer", "testQueue1", true, new Consumer() { @Override public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) { System.out.println("consumerTag = " + consumerTag); System.out.println("messageId = " + basicProperties.getMessageId()); Assertions.assertArrayEquals("hello".getBytes(),body); } }); Assertions.assertTrue(ok); Thread.sleep(500); } @Test public void testBasicAck() throws InterruptedException { boolean ok = virtualHost.exchangeDeclare("testExchange", ExchangeType.DIRECT, true,false,null); Assertions.assertTrue(ok); ok = virtualHost.queueDeclare("testQueue",true, false,false,null); Assertions.assertTrue(ok); // 先发送消息 ok = virtualHost.basicPublish("testExchange","testQueue",null, "hello".getBytes()); Assertions.assertTrue(ok); // 再订阅队列 【要改的地方，把 autoAck 改成 false】 ok = virtualHost.basicConsume("testConsumerTag", "testQueue", false, new Consumer() { @Override public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) { // 消费者自身设定的回调方法 System.out.println("messageId = " + basicProperties.getMessageId()); System.out.println("body = " + new String(body,0,body.length)); Assertions.assertEquals("testQueue",basicProperties.getRoutingKey()); Assertions.assertEquals(1,basicProperties.getDeliverMode()); Assertions.assertArrayEquals("hello".getBytes(),body); // 【要改的地方，新增手动调用 basicAck】 boolean ok = virtualHost.basicAck("testQueue",basicProperties.getMessageId()); Assertions.assertTrue(ok); } }); Assertions.assertTrue(ok); Thread.sleep(500); } }
服务器托管，北京服务器托管，服务器租用 http://www.fwqtg.net
