1、安装 RocketMq

docker search rocketmq

docker pull rocketmqinc/rocketmq:4.4.0  

# 建立 name server
docker run -d -p 9876:9876 -v /data/rocketmq/namesrv/logs:/root/logs -v /data/rocketmq/namesrv/store:/root/store --name rmqnamesrv -e "MAX_POSSIBLE_HEAP=100000000" rocketmqinc/rocketmq:4.4.0 sh mqnamesrv  

# 建立 broker.conf
建立目录和broker配置文件: /data/rocketmq/conf/broker.conf

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
brokerIP1 = 210.112.133.103

# 建立 broker
docker run -d -p 10911:10911 -p 10909:10909 -v  /data/rocketmq/broker/logs:/root/logs -v  /data/rocketmq/broker/store:/root/store -v  /data/rocketmq/conf/broker.conf:/opt/rocketmq-4.4.0/conf/broker.conf  --name rmqbroker --link rmqnamesrv:namesrv  -e "NAMESRV_ADDR=namesrv:9876" -e "MAX_POSSIBLE_HEAP=200000000" rocketmqinc/rocketmq:4.4.0 sh mqbroker -c /opt/rocketmq-4.4.0/conf/broker.conf  

# 安装控制台
docker pull styletang/rocketmq-console-ng  
docker run -e "JAVA_OPTS=-Drocketmq.config.namesrvAddr=210.112.133.103:9876 -Drocketmq.config.isVIPChannel=false" -p 8001:8080 -t styletang/rocketmq-console-ng  

2、POM 加入 starter

       <!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-spring-boot-starter -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.1.1</version>
        </dependency>

3、配置 rocketmq

# rocketmq configuration
rocketmq.name-server=210.112.133.103:9876
rocketmq.producer.group=rocketmq_producer_group
rocketmq.producer.send-message-timeout=3000

4、测试代码

@ApiModel(description= "订单实体")
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Order implements Serializable {

    @ApiModelProperty(value = "主键")
    private Long id;
    @ApiModelProperty(value = "商品名称")
    private String productName;
    @ApiModelProperty(value = "价格")
    private BigDecimal amount;
    @ApiModelProperty(value = "时间")
    private Date createTime;
}


/**
 * 消息生产者
 */
@Service
public class RocketMqProducer {

    private RocketMQTemplate rocketMQTemplate;

    public RocketMqProducer(RocketMQTemplate rocketMQTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
    }

    public RocketMQTemplate getRocketMQTemplate() {
        return rocketMQTemplate;
    }
}


@Api(value = "订单管理", tags = "管理订单增删改查")
@Log4j2
@RestController
@RequestMapping("/order")
public class OrderController {

    private RocketMqProducer rocketMqProducer;

    public OrderController(RocketMqProducer rocketMqProducer) {
        this.rocketMqProducer = rocketMqProducer;
    }

    @ApiOperation(value = "保存或更新订单", notes = "id、productName、amount都是必输入项")
    @ApiImplicitParams(value = {
            @ApiImplicitParam(name = "id", value = "", dataTypeClass = Long.class, required = true),
            @ApiImplicitParam(name = "productName", value = "", dataTypeClass = String.class, required = true),
            @ApiImplicitParam(name = "amount", value = "", dataTypeClass = BigDecimal.class, required = true)
    })
    @PostMapping("/saveOrUpdate")
    public ResponseEntity<Result<Order>> saveOrUpdate(
            @RequestParam Long id,
            @RequestParam String productName,
            @RequestParam BigDecimal amount
    ) {
        Order order = new Order(id, productName, amount, new Date());
        rocketMqProducer.getRocketMQTemplate().getProducer().setSendMsgTimeout(10000);
        rocketMqProducer.getRocketMQTemplate().convertAndSend(Constants.ROCKET_MQ_ORDER, order);
        return ResultUtils.ok(order);
    }
}

@Log4j2
@Service
@RocketMQMessageListener(topic = Constants.ROCKET_MQ_ORDER, selectorExpression = "*", consumerGroup = Constants.ROCKET_MQ_CONSUMER_GROUP_ORDER)
public class RocketMqOrderConsumerListener implements RocketMQListener<Order> {
    @Override
    public void onMessage(Order order) {
        log.error("【onMessage start】");
        log.error(order.getId());
        log.error(order.getProductName());
        log.error(order.getAmount());
        log.error(order.getCreateTime());
        log.error("【onMessage end】");
    }
}

5、控制台