【RelayMQ】基于 Java 实现轻量级消息队列（八）
·
目录
本篇主要介绍消息的发送到生产者消费流程, 以及两种应答方式
一. 消息发送流程
1.1 消息的发送
将消息发送到队列的流程:
- 交换机重命名
- RoutingKey格式校验
- 检查交换机是否存在
- 如果不存在, 则抛出异常, 如果存在, 则判断交换机类型, 进行后续的消息发送
- 如果为直连交换机, 队列名就是RoutingKey
- 如果为扇形交换机或者主题交换机, 遍历关联队列的所有绑定关系, 符合的发送消息
public boolean basicPublic(String exchangeName, String routingKey, BasicProperties basicProperties,byte[] body){
try{
// 1.封装交换机名字
exchangeName = virtualName+exchangeName;
// 2.检查routingKey是否合法
if(!router.checkRoutingKey(routingKey)){
throw new MqException("routingKey非法");
}
// 3.检查交换机对象是否存在
Exchange exchange = memoryDataCenter.getExchange(exchangeName);
if(exchange==null){
throw new MqException("交换机对象不存在");
}
// 4.判断交换机类型
if(exchange.getType() == ExchangeType.DIRECT){
//交换机类型为直接交换机（routingKey为队列的名称）
String queueName = virtualName+routingKey;
// 5.构造消息体
Message message = Message.createMessage(routingKey,body,basicProperties);
// 6.检查队列是否存在
MSGQueue queue = memoryDataCenter.getQueue(queueName);
if(queue==null){
throw new MqException("发送的队列不存在");
}
sendMessage(queue,message);
return true;
}else {
//获取交换机下的所有队列的绑定关系
ConcurrentHashMap<String,Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
for(Map.Entry<String,Binding> entry:bindingsMap.entrySet()){
//根据绑定判断队列是否存在
Binding binding = entry.getValue();
MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
if(queue == null){
System.out.println("发送消息时，发现队列不存在");
continue;
}
//队列存在，则构架消息
Message message = Message.createMessage(routingKey,body,basicProperties);
// 根据类型不同选择对应的发送模式
if(!router.route(exchange.getType(),binding,message)){
continue;
}
sendMessage(queue,message);
return true;
}
}
return true;
} catch (MqException | IOException e) {
System.out.println("发送消息失败");
e.printStackTrace();
return false;
}
}具体的发送行为
- 如果是持久化存储, 数据保存在磁盘中
- 数据保存在内存中
- 调用通知消费者方法
private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException {
int deliverMode = message.getBasicProperties().getDeliverMode();
if(deliverMode == 2){
diskDataCenter.sendMessage(queue, message);
}
memoryDataCenter.sendMessage(queue, message);
//通知可以取消息
consumerManager.notifyConsume(queue.getName());
}1.2 通知消费者
创建一个队列, 标记队列中增加了那些消息, 后续一直遍历这个队列, 处理消息
private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();通知消费者就是将刚刚发送的消息, 放在队列中
public void notifyConsume(String queueName){
tokenQueue.add(queueName);
}二. 消息消费流程
2.1 扫描线程启动
在初始化这个类的时候, 就会开启扫描线程
从tokenQueue队列中不断的取出队列, 消费队列中的消息
public ConsumerManager(VirtualHost parent) {
this.parent = parent;
//创建出扫描线程，不断的扫描队列，并消费其中的消息
scannerThread = new Thread(()->{
while (true){
try{
String queueName = tokenQueue.take();
MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
if(queue==null){
throw new MqException("队列名不存在");
}
synchronized (queue){
//这里只会消费一个消息（来一个消息会添加一个队列名，来几个会添加几个队列名）
consumeMessage(queue);
}
} catch (InterruptedException | MqException e) {
e.printStackTrace();
}
}
});
scannerThread.setDaemon(true);
scannerThread.start();
}2.2 订阅消息
添加一个队列的订阅者，队列接收到消息之后，把消息推送给对应的订阅者
public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
queueName = virtualName+queueName;
try{
consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
System.out.println("订阅消息成功");
return true;
} catch (Exception e) {
System.out.println("订阅消息失败");
return false;
}
}2.3 增加消费者
- 队列的校验
- 添加消费者和队列之间的绑定关系
- 将之前的消息发送到队列中
public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
// 1.检查这个队列是否存在
MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
if(queue == null){
throw new MqException("队列不存在");
}
// 2.创建出消费者和队列的绑定关系
ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
synchronized (queue){
// 3.队列中添加这个消费者关系
queue.addConsumerEnv(consumerEnv);
// 4.开始推送之前的和刚添加的队列中消息（干掉之前发送的消息和刚刚发送的消息）
int n = parent.getMemoryDataCenter().getQueueMessageCount(queueName);
for (int i = 0;i<n;i++){
consumeMessage(queue);
}
}
}2.4 消费消息
- 从确认要推送的队列中挑选一个消费者
- 取出第一个消息
- 将消息发送给消费者
消息发送的时候, 为了避免消息的回调失败, 导致的消息丢失, 会先将消息放在未确认消息中
private void consumeMessage(MSGQueue queue) {
// 1.随机选择一个消费者出来（一个队列可以被多个消费者关注，这些消费者谁先开始有一个优先顺序，挑选一个）
ConsumerEnv luckyDog = queue.chooseConsumer();
if(luckyDog==null){
return;
}
// 2.开始推送消息
//取出第一个消息
Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
if(message==null){
return;
}
// 3.这里推送消息的核心：将消息交给线程池，线程池去调用回调函数（具体的消费者想要执行的业务逻辑）
//也就是去向线程池提交消息的回调函数，线程池会自动执行
workPool.submit(()->{
try{
// 1.先把消息放在待确认的集合中，不能直接执行回调函数，如果回调失败，则会出现消息丢失情况
parent.getMemoryDataCenter().insertMessageWaitAck(queue,message);
// 2.执行回调函数
luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(),message.getBasicProperties(),message.getBody());
// 3.如果是自动应答，则直接把消息进行删除即可
//如果是手动应答，调用basicAck方法来处理
if(luckyDog.isAutoAck()){
// 应答表示已经接收到了消息，那么就需要删除硬盘和内存中的数据
if(message.getBasicProperties().getDeliverMode() == 2){
parent.getDiskDataCenter().deleteMessage(queue,message);
}
parent.getMemoryDataCenter().deleteMessageWaitAck(queue.getName(), message.getMessageId());
parent.getMemoryDataCenter().deleteMessage(message.getMessageId());
}
System.out.println("消息被成功消费");
} catch (IOException | ClassNotFoundException | MqException e) {
e.printStackTrace();
}
});
}三. 应答流程
3.1 消息的发送
生产者会调用mq提供的发送消息方法, 实现这个发送消息的行为, 其中涉及不同交换机类型的转发规则
3.2 消息的存储和通知
在发送消息的后, 需要将消息存储在内存和磁盘中, 并且通知mq处理这个消息
将被添加消息的队列名, 存储在队列中, 设置一个扫描线程, 不断扫描队列, 如果有增加会第一时间处理队列中的消息, 具体的处理行为是交给线程池中的线程来处理, 这里更像是一种推送
3.3 消息的自动应答
自动应答是由线程池中的线程进行操作, 为了避免消息的回调失败, 导致的消息丢失, 会先将消息放在未确认消息中, 然后进行发送, 完成自动应答操作
3.4 消息的手动应答
手动应答是指 消费者在成功处理完消息后，显式地向服务器发送确认请求，通知服务器可以删除该消息 。这种机制与自动应答相反，需要消费者代码主动调用确认方法。
public boolean basicAck(String queueName,String messageId){
queueName = virtualName + queueName;
try{
Message message = memoryDataCenter.getMessage(messageId);
if(message==null){
throw new MqException("消息不存在 消息手动应答失败");
}
MSGQueue queue = memoryDataCenter.getQueue(queueName);
if(queue == null){
throw new MqException("队列不存在 消息手动应答失败");
}
// 删除硬盘数据
if(message.getBasicProperties().getDeliverMode()==2){
diskDataCenter.deleteMessage(queue,message);
}
// 删除内存数据
memoryDataCenter.deleteMessage(messageId);
memoryDataCenter.deleteMessageWaitAck(queueName,messageId);
System.out.println("消息手动应答成功");
return true;
} catch (MqException | IOException | ClassNotFoundException e) {
System.out.println("消息手动应答失败");
return false;
}
}更多推荐
23
0- 0
已为社区贡献4条内容
所有评论(0)