一 MQ介绍
        消息队列是一种“先进先出”的数据结构

        其应用场景主要包含以下三个方面：应用解耦流量削峰数据分发

其主要功能有：

        1.灵活可扩展性、

        2.海量消息堆积能力、

        3.支持顺序消息、

        4.多种消息过滤方式、

        5.支持事务消息、

        6.回溯消费等常用功能。

        RocketMQ 核心的四大组件：Name Server、Broker、Producer、Consumer ，每个组件都可以部署成集群模式进行水平扩展。

二整合springcloud，实现生产和消费

微服务分析：

user_service_9001：生产者

order_service_9002：消费者

2.1 生产者微服务服务搭建（user_service_9001）

1 pom文件导入：


    org.apache.rocketmq
    rocketmq-spring-boot-starter
    2.1.0

2 配置文件中添加如下配置

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: producer_group
    retry-times-when-send-failed: 5
    send-message-timeout: 5000
  topic: ACCOUNT_INFO_TOPIC
  txGroup: producer_group_bank1

3 添加Producer.java类，文件信息如下

package com.yty.system.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.lang.reflect.Type;


@Service
@Slf4j
public class Producer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    
    public void sendAndReceive(String topic,String messageData) throws RuntimeException{
        try {
            Message message = MessageBuilder.withPayload(messageData).build();
            String result = rocketMQTemplate.sendAndReceive(topic,message, String.class);
            log.info("Send message result ：",result);
            if (!result.equals("true")){
                throw new RuntimeException("Send message error");
            }
        }catch (Exception e){
            log.error("Send message error:",e);
            throw new RuntimeException("Send message error:",e);
        }
    }

    
    public void sendMessage(String topic,String messageData) {
        rocketMQTemplate.convertAndSend(topic,messageData);
    }

    
    public SendResult syncSend(String topic,String messageData) {
        SendResult sendResult = rocketMQTemplate.syncSend(topic, messageData);
        log.info("发送结果{}", sendResult);
        return sendResult;
    }
}

4 编写controller，如下

package com.yty.system.controller;


import com.alibaba.fastjson.JSONObject;
import com.yty.system.config.Producer;
import com.yty.system.entity.AccountInfo;
import org.apache.rocketmq.client.producer.SendResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;


@RestController
@RequestMapping("/accountInfo")
public class AccountInfoController {

    @Autowired
    private Producer producer;
    
    @GetMapping(value = "/sendMsg")
    public SendResult sendMsg(){
        try {
            // 发送消息体内容
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("username","zhangsan");
            jsonObject.put("sex","男");
            jsonObject.put("realName","张三");
            // 发送主题
            String topic = "send_message_topic"; // 这个主题很重要，消费者端需要按照这个主题获取消息，所以建议配置在配置文件中
            SendResult result = producer.syncSend(topic,jsonObject.toJSONString());
            return result;
        }catch (Exception e){
            e.printStackTrace();
            return null;
        }
    } 
}

5 测试，需要提前启动rocketMQ和可视化页面，这个在前面章节中已涉及，请参考如下连接

https://blog.csdn.net/ytyDaMoTou/article/details/125343297?spm=1001.2014.3001.5501

启动生产者服务

postman测试

ok，信息发送成功，我们去可视化页面查看

说明消息发送成功，至此，生产者搭建成功，并也成功发送消息，接下来搭建消费者服务

2.1 消费者微服务服务搭建（order_service_9002）

1 导包


    org.apache.rocketmq
    rocketmq-spring-boot-starter
    2.1.0

2 配置文件中编写

# RocketMQ配置
rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: order_service
    retry-times-when-send-failed: 6

3 编写配置类TxmsgConsumer.java，内容如下

package com.yty.order.config;

import com.yty.order.service.IAccountInfoService;
import com.yty.order.service.IDeDuplicationService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

// 特别注意，这里topic必须和生产者的一致
@Component
@RocketMQMessageListener(topic = "send_message_topic",consumerGroup = "consumer_txmsg_group_bank2")
@Slf4j
public class TxmsgConsumer implements RocketMQReplyListener {

    @Autowired
    IAccountInfoService accountInfoService;
    @Autowired
    IDeDuplicationService deDuplicationService;

    @Override
    public Object onMessage(Object message) {
        log.info("开始消费消息，消费信息为:{}",message);
        return true;
    }
}

4 启动消费者服务，可看到获取的信息，如下

至此，说明消费者已获取信息，服务搭建完毕

二 Springcloud集成rocketMQ，实现生产者和消费者案例

二 整合springcloud，实现生产和消费 微服务分析： user_service_9001：生产者 order_service_9002：消费者

Java相关栏目本月热门文章

二整合springcloud，实现生产和消费

微服务分析：

user_service_9001：生产者

order_service_9002：消费者