2024.2.19 模拟实现 RabbitMQ —— 虚拟主机设计_rabbitmq虚拟主机

作者：我家自动化 | 2024-07-18 12:39:24

踩

rabbitmq虚拟主机

引言

针对 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<String,Object> 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<String,Object> 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 = " + routingKey);
            }
//            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<String,Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
               for (Map.Entry<String,Binding> entry : bindingsMap.entrySet()) {
//                   1）获取到绑定对象，判定对应的的队列是否存在
                   Binding binding = entry.getValue();
                   MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                   if(queue == null) {
//                       此处无需抛出异常，可能此处有多个这样的队列
//                       希望不要因为一个队列影响到其他队列的消息的传输
                       System.out.println("[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);
    }
}

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop】