1.2、具体实现

前面已经实现了虚拟主机大部分功能以及转发规则的判定也就是说现在消息已经可以通过

转换机

那么接下来要解决的问题就是消费者该如何订阅消息队列如何把消息推送给消费者以及消费者如何描述自己怎么执行任务~

1.2、具体实现

消费者是以队列为维度订阅消息的并且一个队列可以被多个消费者订阅那么一旦队列中有消息这个消息到底因该给谁呢此处就约定消费者之间按照

“轮询”

这里我们就需要定义一个类ConsumerEnv用来描述一个消费者如下

public

autoAck;//通过这个回调来处理收到的消息private

Consumer

List用来包含若干个上述消费者有哪些消费者订阅了当前队列如下图

ListConsumerEnv

ArrayList();//记录当取到了第几个消费者(AtomicInteger

AtomicInteger

{consumerEnvList.add(consumerEnv);}/***

删除订阅者暂时先不考虑*//***

consumerEnvList.size();consumerSeq.getAndIncrement();//

自增return

consumerEnvList.get(index);}VirtualHost

中订阅消息实现

添加一个队列的订阅者当队列收到消息之后就要把消息推送给对应的订阅者*

param

{consumerManager.addConsumer(consumerTag,

queueName,

consumer);System.out.println([VirtualHost]

basicConsume

{e.printStackTrace();System.out.println([VirtualHost]

basicConsume

false;}}1.2.2、消费者描述自己执行任务方式实现思路

当执行订阅消息的时候我们就让消费者自己去实现处理消息的操作消息的内容通过参数传递具体要干啥取决于消费者自己的业务路基最后再让线程池来执行回调函数.

表达式让消费者在订阅消息的时候就可以实现未来收到消息后如何去处理消息的操作.

FunctionalInterface

如果就一个扫描线程既要获取消息又要执行回调这一个线程可能会忙不过来因为消费者给出的回调具体干什么的咱们是不知道的.

一个简单粗暴的办法就是直接让扫描线程不停的循环遍历所有队列发现有元素就立即处理。

另一个更优雅的办法我采取的办法就是用一个阻塞队列队列中的元素就是接收消息的队列的名字扫描线程只需要盯住这一个阻塞对垒即可此时阻塞队列中传递的队列名就相当于

“令牌”

Executors.newFixedThreadPool(4);//存放令牌的队列private

BlockingQueueString

LinkedBlockingQueue();//扫描线程private

Thread

parent;//创建扫描线程取队列中消费消息scannerThread

new

tokenQueue.take();//2.根据令牌找到队列MSGQueue

queue

parent.getMemoryDataCenter().getQueue(queueName);if(queue

null)

queueName);}//3.从这个队列中消费一个消息synchronized

(queue)

RuntimeException(e);}}});//设置为后台线程scannerThread.setDaemon(true);scannerThread.start();}public

void

{tokenQueue.put(queueName);}/***

添加消费者*

parent.getMemoryDataCenter().getQueue(queueName);if(queue

null)

{queue.addConsumerEnv(consumerEnv);//如果当前队列中已经有一些消息了需要立即消费掉int

parent.getMemoryDataCenter().getMessageCount(queueName);for(int

{//这个方法调用一次就消费一条消息consumeMessage(queue);}}}/***

param

queue.chooseConsumer();if(luckDog

null)

{//当前队列中没有消费者暂时不用消费等后面有消费者了再说return;}//2.从队列中取出一个消息Message

message

parent.getMemoryDataCenter().pollMessage(queue.getName());if(message

null)

{//当前队列中还没有消息也不需要消费return;}//3.把消息带入到消费者的回调方法中丢给线程池执行workerPool.submit(()

{try

{//1.把消息放到待确认的集合当中这个操作一定要在执行回调之前(防止执行回调过程中出现异常导致消息丢失)parent.getMemoryDataCenter().addMessageWaitAck(luckDog.getQueueName(),

message);//2.真正执行回调操作luckDog.getConsumer().handlerDelivery(luckDog.getConsumerTag(),

message.getBasicProperties(),message.getBody());//3.如果当前是

”自动应答“

删除硬盘上的消息if(message.getDeliverMode()

{parent.getDiskDataCenter().deleteMessage(queue,

message);}//2)

删除上面的待确认集合中的消息parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(),

删除内存上的消息中心的消息parent.getMemoryDataCenter().removeMessage(message.getMessageId());System.out.println([ConsumerManager]

消息被成功消费queueName

正确消费就是指消费者的回调方法顺利执行完了没有抛异常之类的这条消息的使命就完成了此时就可以删除了。

在真正执行回调之前把消息放到

autoAckfalse手动应答需要消费者这边在自己的回调方法内部显式调用

basicAck

memoryDataCenter.getQueue(queueName);if(queue

null)

memoryDataCenter.getMessage(messageId);if(message

null)

messageId);}//2.各个维度删除消息if(message.getDeliverMode()

{diskDataCenter.deleteMessage(queue,

message);}memoryDataCenter.removeMessage(messageId);memoryDataCenter.removeMessageWaitAck(queueName,

messageId);System.out.println([VirtualHost]

basicAck

{System.out.println([VirtualHost]

basicAck

messageId);e.printStackTrace();return

false;}}扫描线程完成自动应答

queue.chooseConsumer();if(luckDog

null)

{//当前队列中没有消费者暂时不用消费等后面有消费者了再说return;}//2.从队列中取出一个消息Message

message

parent.getMemoryDataCenter().pollMessage(queue.getName());if(message

null)

{//当前队列中还没有消息也不需要消费return;}//3.把消息带入到消费者的回调方法中丢给线程池执行workerPool.submit(()

{try

{//1.把消息放到待确认的集合当中这个操作一定要在执行回调之前(防止执行回调过程中出现异常导致消息丢失)parent.getMemoryDataCenter().addMessageWaitAck(luckDog.getQueueName(),

message);//2.真正执行回调操作luckDog.getConsumer().handlerDelivery(luckDog.getConsumerTag(),

message.getBasicProperties(),message.getBody());//3.如果当前是

”自动应答“

删除硬盘上的消息if(message.getDeliverMode()

{parent.getDiskDataCenter().deleteMessage(queue,

message);}//2)

删除上面的待确认集合中的消息parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(),

删除内存上的消息中心的消息parent.getMemoryDataCenter().removeMessage(message.getMessageId());System.out.println([ConsumerManager]

消息被成功消费queueName

{e.printStackTrace();}});}如果在回调方法中抛异常了

回调方法中抛异常了后续逻辑执行不到这个消息就会始终呆在待确认的集合中

RabbitMQ

中不叫线程人家是叫进程但是注意这个进程不是操作系统中的进程而是

erlang

待确认集合中每个消息待了多久了如果超出了一定的时间范围就会把这个消息放到一个特定的队列

server

重启之后这个消息就又被加载回内存了就像从来没有被消费过一样消费者就又机会重新拿到这个消息重新消费重复消费的问题是由消费者的业务代码负责保证的broker

server