RabbitMQ使用总结及实例_rabbitmq 本地未启动队列为啥会有

作者：繁依Fanyi0 | 2024-06-08 04:51:54

踩

rabbitmq 本地未启动队列为啥会有

RabbitMQ学习总结

一、基本原理

订阅发布机制,生产者发布消息到队列，消费者订阅该队列，内部的监听机制监听到订阅的队列有消息，就会调用相关的方法进行处理;

二、组件：

生产者、消费者、服务器Server（vhost虚拟主机）、exchange交换机、queue队列、routing key路由键、binding key绑定键

三、各个组件之间关系

生产者发送消息带有routing key的消息头，交换机和队列通过bandging key进行绑定，消息的消息头和banding key相匹配决定消息路由到哪个队列中;
RabbitMQ 的交换机有四种类型：fanout、direct、topic、headers。
direct：交换机的binding key 和消息的routing key相同；
topic：交换机的binding key 和消息的rounting key匹配如：.orange. Routing key为三个单词，且中间的单词为 orange ;

四、简单代码

1、初始化rabbitMQ配置，创建交换机、队列及绑定交换机队列

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitMQ配置
 */
@Slf4j
@Configuration
public class RabbitMQConfig {

    /**
     * 用户服务统一使用的交换机
     */
    public static final String EXCHANGE_ADMIN = "exchange.admin";

    /**
     * 接收用户访问页面的消息，由admin-service服务统一处理
     */
    public static final String QUEUE_ADMIN_VISIT = "admin.visit";
    public static final String ROUTING_KEY_ADMIN_VISIT = "admin.visit";

    @Bean
    public Exchange topicExchangeAdmin() {
        log.info("【业务交换机 " + EXCHANGE_ADMIN + " 创建成功】");
        return ExchangeBuilder.topicExchange(EXCHANGE_ADMIN).durable(true).build();
    }

    /**
     * 创建队列 QUEUE_ADMIN_VISIT
     *
     * @return
     */
    @Bean
    public Queue queueAdminVisit() {
        log.info("【业务队列 " + QUEUE_ADMIN_VISIT + " 创建成功】");
        return QueueBuilder.durable(QUEUE_ADMIN_VISIT).build();
    }

    /**
     * 声明交换机绑定队列 EXCHANGE_ADMIN ->【ROUTING_KEY_ADMIN_VISIT】-> QUEUE_ADMIN_VISIT
     *
     * @return
     */
    @Bean
    public Binding bindingExchangeAdminVisit() {
        log.info("【业务交换机 " + EXCHANGE_ADMIN + " 与业务队列 " + QUEUE_ADMIN_VISIT + " 绑定成功】");
        return BindingBuilder.bind(queueAdminVisit()).to(topicExchangeAdmin()).with(ROUTING_KEY_ADMIN_VISIT).noargs();
    }


    //----------------------------我的统计数据发生变更后的消息队列，用于 advisory服务---------------------------------
    /**
     * 我的统计信息 交换机
     */
    public static final String EXCHANGE_MY_STATISTICS = "exchange.myStatistics";

    /**
     * 我的统计信息 --消息的路由key
     */
    public static final String ROUTING_KEY_STATISTICS_ADVISORY_MODIFY = "myStatistics.advisory.modify";

    /**
     * 我的统计信息 --消息队列
     */
    public static final String QUEUE_STATISTICS_ADVISORY_MODIFY = "myStatistics.advisory.modify";

    /**
     * 实例化交换机
     *
     * @return
     */
    @Bean
    public Exchange topicExchangeMyStatistics() {
        log.info("【业务交换机 " + EXCHANGE_MY_STATISTICS + " 创建成功】");
        return ExchangeBuilder.topicExchange(EXCHANGE_MY_STATISTICS).durable(true).build();
    }

    /**
     * 创建队列 QUEUE_PRODUCT_MODIFY
     *
     * @return
     */
    @Bean
    public Queue queueStatisticsAdvisoryModify() {
        log.info("【业务队列 " + QUEUE_STATISTICS_ADVISORY_MODIFY + " 创建成功】");
        return QueueBuilder.durable(QUEUE_STATISTICS_ADVISORY_MODIFY).build();
    }

    /**
     * 声明交换机绑定队列
     *
     * @return
     */
    @Bean
    public Binding bindingExchangeStatisticsAdvisoryModify() {
        log.info("【业务交换机 " + EXCHANGE_MY_STATISTICS + " 与业务队列 " + QUEUE_STATISTICS_ADVISORY_MODIFY + " 绑定成功】");
        return BindingBuilder.bind(queueStatisticsAdvisoryModify()).to(topicExchangeMyStatistics()).with(ROUTING_KEY_STATISTICS_ADVISORY_MODIFY).noargs();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

2、生产者发布消息


import java.util.Date;

@Service
@Slf4j
public class VisitResRecordService {

    @Autowired
    private IDGenerator idGenerator;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送访问记录消息到MQ中
     *
     * @param visitResourceVo
     */
    public void sendVisitResRecordMessage(VisitResourceVo visitResourceVo) {
        if (null == visitResourceVo) {
            return;
        }

        User user = UserContextHolder.currentUser();

        //由于某些页面特别长，如访问cms的视频时，会导致MQ接收消息的字段不够长，造成报错，所以这里进行拦截
        if (visitResourceVo.getRequestUrl() != null
                && visitResourceVo.getRequestUrl().length() > 200) {

            visitResourceVo.setRequestUrl(visitResourceVo.getRequestUrl().substring(0, 200));
        }

        if (!StringUtils.isEmpty(visitResourceVo.getInvitationCode())) {
            //前端传入的邀请码是加密后的，所以这里需要进行解码
            try {
                visitResourceVo.setInvitationCode(DesHelper.decrypt(visitResourceVo.getInvitationCode()));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        VisitRecordRecordMessage visitRecordRecordMessage = null;
        //构造消息对象，根据当前请求中是否有客户信息进行判断
        if (user != null) {
            visitRecordRecordMessage = VisitRecordRecordMessage.builder()
                    .userIdentifyId(visitResourceVo.getUserIdentifyId())
                    .userToken(visitResourceVo.getUserToken())
                    .userType(visitResourceVo.getUserType())
                    .userId(user.getUserId())
                    .userName(user.getUserName())
                    .userLoginType(user.getUserLoginType())
                    .visitDate(new Date())
                    .resourceType(visitResourceVo.getResourceType())
                    .requestUrl(visitResourceVo.getRequestUrl())
                    .requestParams(visitResourceVo.getRequestParams())
                    .previousPageUrl(visitResourceVo.getPreviousPageUrl())
                    .weixinOpenId(visitResourceVo.getWeixinOpenId())
                    .invitationCode(visitResourceVo.getInvitationCode())
                    .iamSource(visitResourceVo.getIamSource())
                    .build();
        } else {
            visitRecordRecordMessage = VisitRecordRecordMessage.builder()
                    .userIdentifyId(visitResourceVo.getUserIdentifyId())
                    .userToken(visitResourceVo.getUserToken())
                    .userType(visitResourceVo.getUserType())
                    .visitDate(new Date())
                    .resourceType(visitResourceVo.getResourceType())
                    .requestUrl(visitResourceVo.getRequestUrl())
                    .requestParams(visitResourceVo.getRequestParams())
                    .previousPageUrl(visitResourceVo.getPreviousPageUrl())
                    .weixinOpenId(visitResourceVo.getWeixinOpenId())
                    .invitationCode(visitResourceVo.getInvitationCode())
                    .iamSource(visitResourceVo.getIamSource())
                    .build();
        }

        long id = idGenerator.nextId();
        MessageObject messageObject = MessageObject.builder().id(id)
                .messageTypeEnum(MessageTypeEnum.VisitResourceRecord)
                .messageBase(visitRecordRecordMessage)
                .user(UserContextHolder.currentUser())
                .company(CompanyContextHolder.currentCompany())
                .dept(DeptContextHolder.currentDept()).build();

        //为消息设置消息ID，以便在回调时通过该ID来判断是对哪个消息的回调
        CorrelationData correlationData = new CorrelationData(String.valueOf(id));
        log.info("访问资源记录，消息发送到交换机:{} 消息id:{}, msg:{}", RabbitMQConfig.EXCHANGE_ADMIN, correlationData.getId(), messageObject);

        //发送消息
        this.rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_ADMIN,
                RabbitMQConfig.ROUTING_KEY_ADMIN_VISIT,
                JSON.toJSONString(messageObject,
                        SerializerFeature.WriteMapNullValue,
                        SerializerFeature.WriteClassName,
                        SerializerFeature.DisableCircularReferenceDetect), correlationData);
    }

    /**
     * 接收MQ中的访问记录消息，往将数据发给ES，后续客户访问的数据全部从ES中获取
     */
    public void receiveVisitResRecordMessage() {

    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

3、消费者处理消息监听队列，进行消息处理


/**
 * 接收商城订单服务的请求，根据健康服务商品的数量生成工单
 */
@Slf4j
@Component
@RabbitListener(queues = {RabbitMQConfig.QUEUE_ADMIN_VISIT,
                          RabbitMQConfig.QUEUE_PRODUCT_APPROVE,
                          RabbitMQConfig.QUEUE_PRODUCT_UNPUBLISH,
                          RabbitMQConfig.QUEUE_PRODUCT_PUBLISH,
                          RabbitMQConfig.QUEUE_PRODUCT_RECEIVE,
                          RabbitMQConfig.QUEUE_CUSTOMER_REWARD,
                          RabbitMQConfig.QUEUE_CUSTOMER_CPR_REGIST_EXTERNAL,
                          RabbitMQConfig.QUEUE_CUSTOMER_GIVE_VOTE_CHANCE
        })
public class ReceiveMessageListener {

    @Autowired
    private MessageQueueProcessMainService messageQueueProcessMainService;

    @Autowired
    private MessageQueueProcessService messageQueueProcessService;

    @Autowired
    private MessageQueueAbnormalService messageQueueAbnormalService;

    @Autowired
    private IDGenerator idGenerator;

    /**
     * 处理异常消息的线程池
     */
    private static ThreadPoolExecutor threadPoolExecutorForAbnormalMessage;

    static {
        threadPoolExecutorForAbnormalMessage = new ThreadPoolExecutor(
                //线程池维护线程的最少数量
                2,
                //线程池维护线程的最大数量
                10,
                //线程池维护线程所允许的空闲时间
                120, TimeUnit.SECONDS,
                //线程池所使用的缓冲队列
                new ArrayBlockingQueue<Runnable>(100),
                //加入失败，则在调用的主线程上执行
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    /**
     * 监听MQ，消息推送至APP端（采用第三方友盟的服务推送）
     *
     * @param messageObj
     * @param deliveryTag
     * @param channel
     * @throws IOException
     */
    @RabbitHandler
    public void process(@Payload Object messageObj, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {
        try {
            if (messageObj instanceof Message) {
                //获取MQ中的完整消息
                String messageStr = new String(((Message) messageObj).getBody(), "utf-8");
                MessageObject myMessageObject = JSONObject.parseObject(messageStr, new TypeReference<MessageObject>() {
                });

                //获取MQ中完整消息的 业务消息
                JSONObject messageJsonObject = JSONObject.parseObject(messageStr);
                String messageBaseStr = messageJsonObject.get("messageBase").toString();

                log.info("消息id: " + myMessageObject.getId());
                log.info("用户: " + myMessageObject.getUser());
                log.info("收到消息: " + messageBaseStr);
                log.info("完整的消息: " + myMessageObject);

                long id = idGenerator.nextId();

                //对于可能存在大量用户触发且不重要的消息，不用通过DB进行记录和控制
                if (MessageQueueUtils.isNeedRecordInDB(myMessageObject)) {
                    //插入一笔状态为未处理的记录
                    int inserted = messageQueueProcessService.insert(id
                            , myMessageObject.getId(), myMessageObject.getMessageTypeEnum()
                            , messageBaseStr);

                    //判断这个消息之前是否已经处理过
                    if (inserted <= 0) {
                        //由于消息已经存在，需要通过状态确认消息是否处理完成
                        MessageQueueProcess messageQueueProcess = messageQueueProcessService.findByMqId(myMessageObject.getId());
                        //如果是消息已处理，则直接返回
                        if (MessageQueueProcessConstants.Processed.equals(messageQueueProcess.getStatus())) {
                            log.info("----重复消息，不用处理----" + id);
                            channel.basicAck(deliveryTag, false);
                            return;
                        }
                        id = messageQueueProcess.getId();
                    }
                }

                //设置用户上下文，解决服务间RPC调用的权限问题
                UserContextHolder.set(myMessageObject.getUser());
                CompanyContextHolder.set(myMessageObject.getCompany());
                DeptContextHolder.set(myMessageObject.getDept());

                //由于需要使用DB自带的行锁，需要在之前把要锁的数据插入到DB中，否则会引起表锁
                boolean process = messageQueueProcessMainService.process(id, myMessageObject, messageBaseStr);

                channel.basicAck(deliveryTag, false);

                if (process) {
                    log.info("消息处理完成 " + id);
                }
            }
        } catch (Exception e) {
            log.error("消息处理出错 " + e.getMessage());

            try {
                //异常消息放入队列
                if (messageObj instanceof Message) {
                    //获取MQ中的完整消息
                    String messageStr = new String(((Message) messageObj).getBody(), "utf-8");
                    //获取消息对象
                    MessageObject myMessageObject = JSONObject.parseObject(messageStr, new TypeReference<MessageObject>() {
                    });
                    //对于可能存在大量用户触发且不重要的消息，不用通过DB进行记录和控制
                    if (!MessageQueueUtils.isNeedRecordInDB(myMessageObject)) {
                        return;
                    }

                    //获取消息id
                    Long mqId = myMessageObject.getId();

                    //判断异常消息是否已经在DB中，如果不存在则insert
                    boolean isExists = messageQueueAbnormalService.checkIsExists(mqId);
                    if (!isExists) {
                        //异常消息存入DB
                        messageQueueAbnormalService.insert(idGenerator.nextId(), mqId,
                                myMessageObject.getMessageTypeEnum(), messageStr);
                    }
                }
            } catch (Exception ex) {
                ex.printStackTrace();
                log.error("异常消息记录失败 " + e.getMessage());
            }

            //发现异常后，通过新开辟的线程，将消息重新入队
            //通过线程睡眠的方式，避免造成消息不断重复投递的死循环，造成CPU资源浪费，同时可以避免不断打印日志，防止日志文件撑爆服务器
            threadPoolExecutorForAbnormalMessage.execute(() -> {
                try {
                    //10秒后重新入MQ，防止不断的循环
                    Thread.sleep(10 * 1000);

                    //消息重新放入队列
                    channel.basicNack(deliveryTag, false, true);
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            });
        }
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161

4、具体业务处理

/**
 * 队列消息处理主服务
 */
@Service
@Slf4j
public class MessageQueueProcessMainService {
    @Autowired
    private MessageQueueProcessService messageQueueProcessService;

    @Autowired
    private MessageQueueAbnormalService messageQueueAbnormalService;

    @Autowired
    private VisitRecordForESService visitRecordForESService;

    @Autowired
    private OrganizationService organizationService;

    @Autowired
    private RewordService rewordService;

    @Autowired
    private CprService cprService;

    @Autowired
    private MarketingCompetitionService marketingCompetitionService;


    @Transactional(rollbackFor = Exception.class)
    public boolean process(Long id, MessageObject myMessageObject, String messageBase) {
        //对于可能存在大量用户触发且不重要的消息，不用通过DB进行记录和控制
        if (MessageQueueUtils.isNeedRecordInDB(myMessageObject)) {
            log.info("锁定记录ID：" + id);

            //通过DB的行锁，进行加锁处理（事物完成或回滚后会自动释放锁）
            messageQueueProcessService.lockById(id);

            //再次判断是否处理成功，使用双重判断加锁机制
            MessageQueueProcess messageQueueProcess = messageQueueProcessService.findById(id);
            if (MessageQueueProcessConstants.Processed.equals(messageQueueProcess.getStatus())) {
                log.info("----重复消息，不用处理----" + id);
                return false;
            }
        }

        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        log.info("MessageQueueProcessMainService.process 方法中，内嵌业务逻辑 开始处理时间 " + simpleDateFormat.format(new Date()));

        //处理页面访问资源的消息
        if (MessageTypeEnum.VisitResourceRecord.equals(myMessageObject.getMessageTypeEnum())) {
            VisitRecordRecordMessage visitRecordRecordMessage = JSONObject.parseObject(messageBase, new TypeReference<VisitRecordRecordMessage>() {
            });
            log.info("调用visitRecordForESService.record方法，记录前端访问的页面");
            visitRecordForESService.record(visitRecordRecordMessage);
        } else {
            throw new BaseException("MessageQueueProcessMainService.process方法传入的队列参数值有误");
        }

        log.info("MessageQueueProcessMainService.process 方法中，内嵌业务逻辑 结束处理时间 " + simpleDateFormat.format(new Date()));

        if (MessageQueueUtils.isNeedRecordInDB(myMessageObject)) {
            //更改状态
            messageQueueProcessService.updateStatus2Finish(id);

            //删除消息异常表中对应的记录
            messageQueueAbnormalService.delete(myMessageObject.getId());
        }

        log.info("----完成业务处理----" + id);

        return true;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

5、具体业务处理sql

/**
 * 消息队列辅助类相关的服务
 */
public interface MessageQueueMapper {

    //--------------------------消息处理相关---------------------------------

    /**
     * 消息处理表中插入记录
     *
     * @param messageQueueProcess
     * @return
     */
    @Insert("insert into message_queue_process(id, create_by, create_time, mq_id, mq_type, message_part, status) " +
            " values(#{id}, #{createBy}, #{createTime}, #{mqId}, #{messageTypeEnum}, #{messagePart}, #{status})")
    public int insertMessageQueueProcess(MessageQueueProcess messageQueueProcess);

    /**
     * 通过消息mqId获取记录
     *
     * @param mqId
     * @return
     */
    @Select("select id, status from message_queue_process where mq_id=#{mqId}")
    public MessageQueueProcess findMessageQueueProcessByMqId(Long mqId);

    /**
     * 锁定id所在记录
     *
     * @param id
     * @return
     */
    @Select("select 1 from message_queue_process where id=#{id} for update")
    public MessageQueueProcess lockMessageQueueProcessById(Long id);

    /**
     * 通过id获取消息处理的状态
     *
     * @param id
     * @return
     */
    @Select("select status from message_queue_process where id=#{id}")
    public MessageQueueProcess findMessageQueueProcessById(Long id);

    /**
     * 通过id更新消息处理的状态
     *
     * @param id
     * @param status
     * @return
     */
    @Update("update message_queue_process set status=#{status} where id=#{id}")
    public int updateMessageQueueProcessStatus(Long id, String status);

    //--------------------------异常消息相关---------------------------------

    /**
     * 通过消息mqId获取异常记录
     *
     * @param mqId
     * @return
     */
    @Select("select id from message_queue_abnormal where mq_id=#{mqId}")
    public MessageQueueAbnormal findMessageQueueAbnormalByMqId(Long mqId);

    /**
     * 消息异常表，插入数据
     *
     * @param messageQueueAbnormal
     * @return
     */
    @Insert("insert into message_queue_abnormal(id, create_time, mq_id, mq_type, message_all) " +
            " values(#{id}, now(), #{mqId}, #{messageTypeEnum}, #{messageAll})")
    public int insertMessageQueueAbnormal(MessageQueueAbnormal messageQueueAbnormal);

    /**
     * 根据消息mqId删除消息异常表的记录
     *
     * @param mqId
     * @return
     */
    @Delete("delete from message_queue_abnormal where mq_id = #{mqId}")
    public int deleteMessageQueueAbnormal(Long mqId);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

总结：一、具体业务处理时使用数据库锁进行幂等处理，判断当前消息是否已被处理：

1.消息信息入库（如果已存在并判断状态，已处理入库），且状态为未处理；

2.进行具体业务处理流程，处理前使用数据库显式锁 select 1 from message_queue_process where id=#{id} for update 将对应消息记录加锁;再次判断记录状态（双重判断），为未处理进行具体业务

3.业务处理完成,对应消息记录状态更新为已处理;且删除异常消息表中对应记录（如果存在）

二、异常消息处理：

1.异常消息入异常消息表用于记录；

2.开启线程池将消息重新入消息队列；

其实这个例子里面少了binding key，只是这里将binding key和routing key合为一个;

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/繁依Fanyi0/article/detail/688486