RabbitMQ保证消息被成功发送和消费

发布于:2024-07-11 ⋅ 阅读:(20) ⋅ 点赞:(0)

一 : 在使用 RabbitMQ 作为消息队列时,保证消息被成功发送和消费是一个非常重要的问题。以下是一些关键点和最佳实践,以确保消息的可靠传输和处理。*
配置方式:

保证消息被成功发送
确认模式(Confirm Mode):

生产者可以启用确认模式,确保消息成功到达交换机。
使用 channel.confirmSelect() 启用确认模式。
使用 channel.waitForConfirms() 或 channel.addConfirmListener() 来处理确认消息。
事务模式(Transaction Mode):

生产者可以使用事务模式,确保消息成功到达队列。
使用 channel.txSelect() 开启事务,channel.txCommit() 提交事务,channel.txRollback() 回滚事务。
处理发送失败:

实现重试机制,可以在发送失败时重试。
使用死信交换机(Dead Letter Exchange, DLX)来存储处理失败的消息。
保证消息被成功消费
手动确认(Manual Acknowledgment):

消费者应该使用手动确认模式,确保消息被成功处理后再确认。
使用 channel.basicConsume(queue, false, consumer) 开启手动确认模式。
在消息处理成功后,调用 channel.basicAck(deliveryTag, false) 确认消息。
处理消费失败:

实现消费失败的重试机制。
使用死信交换机(DLX)来存储处理失败的消息。
幂等性:

确保消费者处理消息的幂等性,避免重复消费导致的问题。

二:通过记录消息到数据库中,采用定时任务轮询方式:

1 这是一个 Spring 组件,用于构建和发布返利消息事件。

//topic 字段从配置文件中获取,表示消息队列的 topic。
//buildEventMessage 方法用于构建 EventMessage 对象,包含随机生成的 ID、当前时间戳和数据。
//topic 方法返回消息队列的 topic。
//RebateMessage 是一个内部类,定义了返利消息的结构


@Component
public class SendRebateMessageEvent extends BaseEvent<SendRebateMessageEvent.RebateMessage> {

    @Value("${spring.rabbitmq.topic.send_rebate}")
    private String topic;

    @Override
    public EventMessage<RebateMessage> buildEventMessage(RebateMessage data) {
        return EventMessage.<SendRebateMessageEvent.RebateMessage>builder()
                .id(RandomStringUtils.randomNumeric(11))
                .timestamp(new Date())
                .data(data)
                .build();
    }

    @Override
    public String topic() {
        return topic;
    }

    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public static class RebateMessage {
        private String userId;
        private String rebateDesc;
        private String rebateType;
        private String rebateConfig;
        private String bizId;
    }
}

@Data
public abstract class BaseEvent<T> {

    public abstract EventMessage<T> buildEventMessage(T data);

    public abstract String topic();

    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public static class EventMessage<T> {
        private String id;
        private Date timestamp;
        private T data;
    }

}

2 生产者示例

//topic 字段从配置文件中获取,表示消息队列的 topic。
//listener 方法使用 @RabbitListener 注解监听指定队列的消息。
//消息到达后,解析消息内容,根据 rebateType 字段的不同,调用相应的服务方法处理消息。
//异常处理机制确保了消息处理的健壮性。


@Component
public class RebateMessageCustomer {

    @Value("${spring.rabbitmq.topic.send_rebate}")
    private String topic;
    @Resource
    private IRaffleActivityAccountQuotaService raffleActivityAccountQuotaService;
    @Resource
    private ICreditAdjustService creditAdjustService;

    @RabbitListener(queuesToDeclare = @Queue(value = "${spring.rabbitmq.topic.send_rebate}"))
    public void listener(String message) {
        try {
            log.info("监听用户行为返利消息 topic: {} message: {}", topic, message);
            BaseEvent.EventMessage<SendRebateMessageEvent.RebateMessage> eventMessage = JSON.parseObject(message, new TypeReference<BaseEvent.EventMessage<SendRebateMessageEvent.RebateMessage>>() {
            }.getType());
            SendRebateMessageEvent.RebateMessage rebateMessage = eventMessage.getData();

            switch (rebateMessage.getRebateType()) {
                case "sku":
                    SkuRechargeEntity skuRechargeEntity = new SkuRechargeEntity();
                    skuRechargeEntity.setUserId(rebateMessage.getUserId());
                    skuRechargeEntity.setSku(Long.valueOf(rebateMessage.getRebateConfig()));
                    skuRechargeEntity.setOutBusinessNo(rebateMessage.getBizId());
                    skuRechargeEntity.setOrderTradeType(OrderTradeTypeVO.rebate_no_pay_trade);
                    raffleActivityAccountQuotaService.createOrder(skuRechargeEntity);
                    break;
                case "integral":
                    TradeEntity tradeEntity = new TradeEntity();
                    tradeEntity.setUserId(rebateMessage.getUserId());
                    tradeEntity.setTradeName(TradeNameVO.REBATE);
                    tradeEntity.setTradeType(TradeTypeVO.FORWARD);
                    tradeEntity.setAmount(new BigDecimal(rebateMessage.getRebateConfig()));
                    tradeEntity.setOutBusinessNo(rebateMessage.getBizId());
                    creditAdjustService.createOrder(tradeEntity);
                    break;
            }
        } catch (AppException e) {
            if (ResponseCode.INDEX_DUP.getCode().equals(e.getCode())) {
                log.warn("监听用户行为返利消息,消费重复 topic: {} message: {}", topic, message, e);
                return;
            }
            throw e;
        } catch (Exception e) {
            log.error("监听用户行为返利消息,消费失败 topic: {} message: {}", topic, message, e);
            throw e;
        }
    }
}

3 消费者示例


//这个方法用于保存用户返利记录,并在事务中插入用户行为返利订单和任务对象。
//在事务外,同步发送 MQ 消息。
//发送消息时,调用 eventPublisher.publish 方法发布消息到指定的 topic。


public void saveUserRebateRecord(String userId, List<BehaviorRebateAggregate> behaviorRebateAggregates) {
    try {
        dbRouter.doRouter(userId);
        transactionTemplate.execute(status -> {
            try {
                for (BehaviorRebateAggregate behaviorRebateAggregate : behaviorRebateAggregates) {
                    BehaviorRebateOrderEntity behaviorRebateOrderEntity = behaviorRebateAggregate.getBehaviorRebateOrderEntity();
                    UserBehaviorRebateOrder userBehaviorRebateOrder = new UserBehaviorRebateOrder();
                    userBehaviorRebateOrder.setUserId(behaviorRebateOrderEntity.getUserId());
                    userBehaviorRebateOrder.setOrderId(behaviorRebateOrderEntity.getOrderId());
                    userBehaviorRebateOrder.setBehaviorType(behaviorRebateOrderEntity.getBehaviorType());
                    userBehaviorRebateOrder.setRebateDesc(behaviorRebateOrderEntity.getRebateDesc());
                    userBehaviorRebateOrder.setRebateType(behaviorRebateOrderEntity.getRebateType());
                    userBehaviorRebateOrder.setRebateConfig(behaviorRebateOrderEntity.getRebateConfig());
                    userBehaviorRebateOrder.setOutBusinessNo(behaviorRebateOrderEntity.getOutBusinessNo());
                    userBehaviorRebateOrder.setBizId(behaviorRebateOrderEntity.getBizId());
                    userBehaviorRebateOrderDao.insert(userBehaviorRebateOrder);

                    TaskEntity taskEntity = behaviorRebateAggregate.getTaskEntity();
                    Task task = new Task();
                    task.setUserId(taskEntity.getUserId());
                    task.setTopic(taskEntity.getTopic());
                    task.setMessageId(taskEntity.getMessageId());
                    task.setMessage(JSON.toJSONString(taskEntity.getMessage()));
                    task.setState(taskEntity.getState().getCode());
                    taskDao.insert(task);
                }
                return 1;
            } catch (DuplicateKeyException e) {
                status.setRollbackOnly();
                log.error("写入返利记录,唯一索引冲突 userId: {}", userId, e);
                throw new AppException(ResponseCode.INDEX_DUP.getCode(), ResponseCode.INDEX_DUP.getInfo());
            }
        });
    } finally {
        dbRouter.clear();
    }

    for (BehaviorRebateAggregate behaviorRebateAggregate : behaviorRebateAggregates) {
        TaskEntity taskEntity = behaviorRebateAggregate.getTaskEntity();
        Task task = new Task();
        task.setUserId(taskEntity.getUserId());
        task.setMessageId(taskEntity.getMessageId());
        try {
            eventPublisher.publish(taskEntity.getTopic(), taskEntity.getMessage());
            taskDao.updateTaskSendMessageCompleted(task);
        } catch (Exception e) {
            log.error("写入返利记录,发送MQ消息失败 userId: {} topic: {}", userId, task.getTopic());
            taskDao.updateTaskSendMessageFail(task);
        }
    }
}


public List<String> createOrder(BehaviorEntity behaviorEntity) {
        // 1. 查询返利配置
        List<DailyBehaviorRebateVO> dailyBehaviorRebateVOS = behaviorRebateRepository.queryDailyBehaviorRebateConfig(behaviorEntity.getBehaviorTypeVO());
        if (null == dailyBehaviorRebateVOS || dailyBehaviorRebateVOS.isEmpty()) return new ArrayList<>();

        // 2. 构建聚合对象
        List<String> orderIds = new ArrayList<>();
        List<BehaviorRebateAggregate> behaviorRebateAggregates = new ArrayList<>();
        for (DailyBehaviorRebateVO dailyBehaviorRebateVO : dailyBehaviorRebateVOS) {
            // 拼装业务ID;用户ID_返利类型_外部透彻业务ID
            String bizId = behaviorEntity.getUserId() + Constants.UNDERLINE + dailyBehaviorRebateVO.getRebateType() + Constants.UNDERLINE + behaviorEntity.getOutBusinessNo();
            BehaviorRebateOrderEntity behaviorRebateOrderEntity = BehaviorRebateOrderEntity.builder()
                    .userId(behaviorEntity.getUserId())
                    .orderId(RandomStringUtils.randomNumeric(12))
                    .behaviorType(dailyBehaviorRebateVO.getBehaviorType())
                    .rebateDesc(dailyBehaviorRebateVO.getRebateDesc())
                    .rebateType(dailyBehaviorRebateVO.getRebateType())
                    .rebateConfig(dailyBehaviorRebateVO.getRebateConfig())
                    .outBusinessNo(behaviorEntity.getOutBusinessNo())
                    .bizId(bizId)
                    .build();
            orderIds.add(behaviorRebateOrderEntity.getOrderId());

            // MQ 消息对象
            SendRebateMessageEvent.RebateMessage rebateMessage = SendRebateMessageEvent.RebateMessage.builder()
                    .userId(behaviorEntity.getUserId())
                    .rebateType(dailyBehaviorRebateVO.getRebateType())
                    .rebateConfig(dailyBehaviorRebateVO.getRebateConfig())
                    .bizId(bizId)
                    .build();

            // 构建事件消息
            BaseEvent.EventMessage<SendRebateMessageEvent.RebateMessage> rebateMessageEventMessage = sendRebateMessageEvent.buildEventMessage(rebateMessage);

            // 组装任务对象
            TaskEntity taskEntity = new TaskEntity();
            taskEntity.setUserId(behaviorEntity.getUserId());
            taskEntity.setTopic(sendRebateMessageEvent.topic());
            taskEntity.setMessageId(rebateMessageEventMessage.getId());
            taskEntity.setMessage(rebateMessageEventMessage);
            taskEntity.setState(TaskStateVO.create);

            BehaviorRebateAggregate behaviorRebateAggregate = BehaviorRebateAggregate.builder()
                        .userId(behaviorEntity.getUserId())
                        .behaviorRebateOrderEntity(behaviorRebateOrderEntity)
                        .taskEntity(taskEntity)
                        .build();

            behaviorRebateAggregates.add(behaviorRebateAggregate);
        }

        // 3. 存储聚合对象数据
        behaviorRebateRepository.saveUserRebateRecord(behaviorEntity.getUserId(), behaviorRebateAggregates);

        // 4. 返回订单ID集合
        return orderIds;
    }



@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class TaskEntity {

    /** 活动ID */
    private String userId;
    /** 消息主题 */
    private String topic;
    /** 消息编号 */
    private String messageId;
    /** 消息主体 */
    private BaseEvent.EventMessage<SendRebateMessageEvent.RebateMessage> message;
    /** 任务状态;create-创建、completed-完成、fail-失败 */
    private TaskStateVO state;

}

@Data
public class Task {

    /** 自增ID */
    private String id;
    /** 活动ID */
    private String userId;
    /** 消息主题 */
    private String topic;
    /** 消息编号 */
    private String messageId;
    /** 消息主体 */
    private String message;
    /** 任务状态;create-创建、completed-完成、fail-失败 */
    private String state;
    /** 创建时间 */
    private Date createTime;
    /** 更新时间 */
    private Date updateTime;

}

4 定时任务示例

// @Scheduled(cron = "0/5 * * * * ?")
public void exec_db01() {
    try {
        // 设置库表
        dbRouter.setDBKey(1);
        dbRouter.setTBKey(0);
        // 查询未发送的任务
        List<TaskEntity> taskEntities = taskService.queryNoSendMessageTaskList();
        if (taskEntities.isEmpty()) return;
        // 发送MQ消息
        for (TaskEntity taskEntity : taskEntities) {
            try {
                taskService.sendMessage(taskEntity);
                taskService.updateTaskSendMessageCompleted(taskEntity.getUserId(), taskEntity.getMessageId());
            } catch (Exception e) {
                log.error("定时任务,发送MQ消息失败 userId: {} topic: {}", taskEntity.getUserId(), taskEntity.getTopic());
                taskService.updateTaskSendMessageFail(taskEntity.getUserId(), taskEntity.getMessageId());
            }
        }
    } catch (Exception e) {
        log.error("定时任务,扫描MQ任务表发送消息失败。", e);
    } finally {
        dbRouter.clear();
    }
}


@Data
public class TaskEntity {

    /** 活动ID */
    private String userId;
    /** 消息主题 */
    private String topic;
    /** 消息编号 */
    private String messageId;
    /** 消息主体 */
    private String message;

}

三: 串联流程

生产消息:

saveUserRebateRecord 方法在事务中插入用户行为返利订单和任
务对象。
在事务外,调用 eventPublisher.publish 方法发布消息到指定的 
topic。
消息构建和发布:

SendRebateMessageEvent 类构建 EventMessage 对象,包含随机
生成的 ID、当前时间戳和数据,并返回消息队列的 topic。
消费消息:

RebateMessageCustomer 类监听指定队列的消息,解析消息内容,
根据 rebateType 字段的不同,调用相应的服务方法处理消息。
定时任务补偿:

SendMessageTaskJob 类定时扫描数据库中的任务表,发送未发送的
消息到 MQ 队列,并更新任务状态。如果发送失败,记录错误日志并
更新任务状态为发送失败。

四:配置文件

spring:
  rabbitmq:
    addresses: ****
    port: ***
    username: **
    password: **
    listener:
      simple:
        prefetch: 1 # 每次投递n个消息,消费完在投递n个
    topic:
      send_rebate: send_rebate

五: 消费失败

消息发送:
生产者在发送消息时,会将消息的相关信息(如消息内容、发送状态等)记录到 task 表中。
如果消息发送成功,则更新 task 表中的状态为“已发送”。
如果消息发送失败,则更新 task 表中的状态为“发送失败”。
定时任务会扫描 task 表,查找状态为“发送失败”的消息,并重试发送。

消息消费:
消费者在处理消息时,也会将消息的相关信息(如消息内容、处理状态等)记录到 task 表中。
如果消息处理成功,则更新 task 表中的状态为“已处理”。
如果消息处理失败,则更新 task 表中的状态为“处理失败”。
定时任务会扫描 task 表,查找状态为“处理失败”的消息,并重试处理。
通过这种方式,可以确保即使消息在发送或消费过程中出现失败,也能够通过重试机制最终成功发送或处理。

示例:

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class MessageService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private TaskRepository taskRepository;

    @Transactional
    public void sendMessage(String message) {
        try {
            rabbitTemplate.convertAndSend("send_rebate", message);
            TaskEntity task = new TaskEntity();
            task.setMessage(message);
            task.setStatus("SENT");
            taskRepository.save(task);
        } catch (Exception e) {
            TaskEntity task = new TaskEntity();
            task.setMessage(message);
            task.setStatus("SEND_FAILED");
            taskRepository.save(task);
        }
    }
}

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class MessageConsumer {

    @Autowired
    private TaskRepository taskRepository;

    @RabbitListener(queues = "${spring.rabbitmq.topic.send_rebate}")
    public void handleMessage(String message) {
        try {
            // 处理消息的逻辑
            System.out.println("Processing message: " + message);
            // 假设处理成功
            TaskEntity task = new TaskEntity();
            task.setMessage(message);
            task.setStatus("PROCESSED");
            taskRepository.save(task);
        } catch (Exception e) {
            // 处理失败的逻辑
            TaskEntity task = new TaskEntity();
            task.setMessage(message);
            task.setStatus("PROCESS_FAILED");
            taskRepository.save(task);
        }
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class RetryTaskJob {

    @Autowired
    private TaskRepository taskRepository;

    @Autowired
    private MessageService messageService;

    @Autowired
    private MessageConsumer messageConsumer;

    @Scheduled(fixedRate = 60000) // 每分钟执行一次
    public void retryFailedMessages() {
        // 重试发送失败的消息
        List<TaskEntity> sendFailedTasks = taskRepository.findByStatus("SEND_FAILED");
        for (TaskEntity task : sendFailedTasks) {
            try {
                messageService.sendMessage(task.getMessage());
                task.setStatus("SENT");
                taskRepository.save(task);
            } catch (Exception e) {
                // 记录日志或进行其他处理
            }
        }

        // 重试处理失败的消息
        List<TaskEntity> processFailedTasks = taskRepository.findByStatus("PROCESS_FAILED");
        for (TaskEntity task : processFailedTasks) {
            try {
                messageConsumer.handleMessage(task.getMessage());
                task.setStatus("PROCESSED");
                taskRepository.save(task);
            } catch (Exception e) {
                // 记录日志或进行其他处理
            }
        }
    }
}

六: 防止重复消费

保证消息消费的幂等性是确保消息系统可靠性的重要一环。幂等性意
味着无论消息被处理一次还是多次,结果都是相同的。以下是一些常见的策略来保证消息消费的幂等性:

唯一标识符:为每条消息生成一个唯一的标识符(如 UUID),并在
处理消息时检查该标识符是否已经被处理过。如果已经处理过,则忽略该消息。

状态检查:在处理消息之前,检查系统的状态,确保该消息对应的操
作尚未执行。例如,如果消息是要更新某个资源,可以先检查该资源的状态,确保更新操作尚未执行。

数据库唯一约束:在数据库中为消息处理结果创建唯一约束,确保相
同的消息不会被重复处理。

幂等API设计:设计幂等的API,确保相同的请求多次执行不会产生不
同的结果。例如,使用“PUT”方法更新资源,而不是“POST”方法。

使用 Redis 来实现消息消费的幂等性是一个非常有效的方法。Redis 是一个高性能的内存数据库,适合用于存储临时状态信息。
1 消费者处理消息并记录到 Redis

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class MessageConsumer {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @RabbitListener(queues = "${spring.rabbitmq.topic.send_rebate}")
    public void handleMessage(String message) {
        // 假设消息中包含唯一标识符
        String messageId = extractMessageId(message);

        // 检查消息是否已经处理过或正在处理
        if (redisTemplate.hasKey(messageId)) {
            System.out.println("Message already processed or being processed: " + messageId);
            return;
        }

        // 将消息ID存入Redis,设置过期时间
        redisTemplate.opsForValue().set(messageId, "PROCESSING", 60, TimeUnit.SECONDS);

        try {
            // 处理消息的逻辑
            System.out.println("Processing message: " + message);
            // 假设处理成功
            redisTemplate.opsForValue().set(messageId, "PROCESSED", 60, TimeUnit.SECONDS);
        } catch (Exception e) {
            // 处理失败的逻辑
            redisTemplate.delete(messageId); // 删除Redis中的记录,以便可以重试
        }
    }

    private String extractMessageId(String message) {
        // 假设消息中包含唯一标识符,例如 JSON 格式中的 "id" 字段
        // 这里只是一个示例,实际实现可能需要解析消息内容
        return message.substring(0, 36); // 假设 UUID 长度为 36
    }
}

spring:
  redis:
    host: localhost
    port: 6379

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

通过上述方法和配置,您可以确保消息消费的幂等性。消费者在处理消息之前会检查消息的唯一标识符是否已经存在于 Redis 中,如果存在,则忽略该消息,从而避免重复处理。同时,通过设置过期时间,可以确保在处理过程中出现异常时,Redis 中的记录会被删除,从而允许消息重试。