若依微服务项目整合rocketMq

原文链接：ttps://mp.weixin.qq.com/s/IYdo_suKvvReqCiEKjCeHw
第一步下载若依项目
第二步安装rocketMq（推荐在linux使用docker部署比较快）
第二步新建一个生产者模块儿，再建一个消费者模块

第四步在getway模块中配置接口映射规则
第五步添加依赖（生产者和消费者他们的rocketmq依赖一样）

       <dependency>
            <groupId&gt;org.apache.rocketmq</groupId&gt;
            <artifactId&gt;rocketmq-spring-boot-starter</artifactId&gt;
            # 选择适宜你的版本就好了我的若依项目版本是3.6.3
            <version&gt;2.0.2</version&gt;
        </dependency&gt;

第六步添加配置
生产者的

# Tomcat
server:
  port: 10301

# Spring
spring: 
  application:
    # 应用名称
    name: mq-producer
  profiles:
    # 环境配置
    active: dev
  cloud:
    nacos:
      discovery:
        # 服务注册地址
        server-addr: 127.0.0.1:9849
      config:
        # 配置中心地址
        server-addr: 127.0.0.1:9849
        # 配置文件格式
        file-extension: yml
        # 共享配置
        shared-configs:
          - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
rocketmq:
  name-server: 192.168.94.130:9876
  producer:
    # 发送同一类消息的设置为同一个group，保证唯一
    group: springboot_producer_group
    # 发送消息超时时间，默认3000
    sendMessageTimeout: 10000
    # 发送消息失败重试次数，默认2
    retryTimesWhenSendFailed: 2
    # 异步消息重试此处，默认2
    retryTimesWhenSendAsyncFailed: 2
    # 消息最大长度，默认1024 * 1024 * 4(默认4M)
    maxMessageSize: 4096
    # 压缩消息阈值，默认4k(1024 * 4)
    compressMessageBodyThreshold: 4096
    # 是否在内部发送失败时重试另一个broker，默认false
    retryNextServer: false
#    stream:
#      bindings:
#        ## 新版本固定格式  函数名-{out/in}-{index}
#        demoChannel-out-0:
#          destination: stream-test-topic
#      rocketmq:
#        binder:
#          name-server: 192.168.94.130:9876

消费者的

# Tomcat
server:
  port: 10302

# Spring
spring: 
  application:
    # 应用名称
    name: mq-consumer
  profiles:
    # 环境配置
    active: dev
  cloud:
    nacos:
      discovery:
        # 服务注册地址
        server-addr: 127.0.0.1:9849
      config:
        # 配置中心地址
        server-addr: 127.0.0.1:9849
        # 配置文件格式
        file-extension: yml
        # 共享配置
        shared-configs:
          - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
rocketmq:
  consumer:
    group: springboot_consumer_group
    # 一次拉取消息最大值，注意是拉取消息的最大值而非消费最大值
    pull-batch-size: 10
  name-server: 192.168.94.130:9876

逻辑代码：
生产消息的代码

package com.zhj.mqproducer.controller;

import lombok.Setter;
import org.apache.rocketmq.client.producer.SendResult;
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.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@RestController
@RequestMapping("/mqpro")
public class SendMessageController {


    @Setter(onMethod_ = @Autowired)
    private RocketMQTemplate rocketmqTemplate;

    @GetMapping("/test")
    public  String test() {
        Message<String> msg = MessageBuilder.withPayload("Hello,RocketMQ").build();
        rocketmqTemplate.send("eatfan", msg);
        return "haha";
    }

}

消费消息的代码

package com.zhj.mqconsumer.domain;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(topic = "eatfan", consumerGroup = "consumer-zhj2")
public class MyConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Received message: " + message);
    }
}

在浏览器地址中输入如图信息发送消息

控制台查看输出

代码007普通

打赏 收藏 海报 链接