spring-cloud-starter-rocketmq

mvn clean install

<dependency>
    <groupId>com.uhu</groupId>
    <artifactId>spring-cloud-starter-rocketmq</artifactId>
    <version>1.0-SNAPSHOT</version>
</dependency>

rocketmq:
  serverIp: 192.168.191.129
  port: 9876
  consumer-group: mygroup
  send-timeout: 7000

package com.uhu.cloud.demo2.biz.mq;

import com.uhu.cloud.demo2.biz.dal.dataobject.Course;
import com.uhu.rocketmq.manager.RocketManager;
import com.uhu.rocketmq.message.SimpleMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @Author Bomber
 * @Description 消息消费者
 * @Date 2023/5/11 16:29
 * @Version 1.0
 */

@Component
public class Consumers {

    @Autowired
    private RocketManager rocketManager;

    @PostConstruct
    public void init() {
        /* 
        * 方式1：创建一个消息主题，且进行消费
        *
        * 参数1：主题
        * 参数2：标签表达式，默认为*
        * 参数3：消费者
        * */
        System.out.println("创建【checkCourse】主题");
        rocketManager.createTopic("checkCourse", null, (msg) -> {
            System.out.println("选择了课程：" + SimpleMessage.getContent(msg, Course.class));
        });
        
        /*
         * 方式2：创建一个消息主题，且仅作为生产
         * 
         * 参数1：主题
         * */
        System.out.println("创建【checkOther】主题");
        rocketManager.createTopic("checkOther");
    }
}

public class Controller {
    //注入bean
    @Autowired
    private MessageSender messageSender;

    public void sendMessage() {
        // 创建消息
        SimpleMessage<Course> courseSimpleMessage = new SimpleMessage<>("checkCourse", "tag1", new Course(1L, "语文课"));

        // 发送消息
        messageSender.sendMessage(courseSimpleMessage);
    }
}

package com.uhu.cloud.demo2.biz.dal.dataobject;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @Author Bomber
 * @Description 课程
 * @Date 2023/5/11 16:34
 * @Version 1.0
 */

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Course {
    private Long id;
    private String name;
}

package com.uhu.cloud.demo2.biz.mq;

import com.uhu.cloud.demo2.biz.dal.dataobject.Course;
import com.uhu.rocketmq.manager.RocketManager;
import com.uhu.rocketmq.message.SimpleMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @Author Bomber
 * @Description 消息消费者
 * @Date 2023/5/11 16:29
 * @Version 1.0
 */

@Component
public class Consumers {

    @Autowired
    private RocketManager rocketManager;

    @PostConstruct
    public void init() {
        // 创建一个消息课程，且进行打印输出消费
        /*
        * 参数1：主题
        * 参数2：标签表达式，默认为*
        * 参数3：消费者
        * */
        System.out.println("创建【checkCourse】主题");
        rocketManager.createTopic("checkCourse", null, (msg) -> {
            System.out.println("选择了课程：" + SimpleMessage.getContent(msg, Course.class));
        });
    }
}

@GetMapping("/send")
public void send() {

    // ...中间业务省略
	// 编辑消息简单消息（内部封装课程对象）
	SimpleMessage<Course> courseSimpleMessage = new SimpleMessage<>("checkCourse", "tag1", new Course(1L, "语文课"));
    
    // 发送简单消息
	messageSender.sendMessage(courseSimpleMessage);
}

package com.uhu.rocketmq.message;

import com.alibaba.fastjson2.JSONObject;
import org.apache.rocketmq.common.message.Message;

import java.nio.charset.StandardCharsets;

/**
 * @Author Bomber
 * @Description 简单消息
 * @Date 2023/5/11 16:17
 * @Version 1.0
 */
public class SimpleMessage<T> extends Message {

    /**
     * 简单消息
     * @param topic 主题
     * @param tags 标签
     * @param content 内容
     */
    public SimpleMessage(String topic, String tags, T content) {
        
    }

    /**
     * 获取消息内容对象
     * @param msg 消息
     * @param contentClass 消息内容类型
     * @param <T>
     * @return
     */
    public static <T> T getContent(Message msg, Class<T> contentClass) {
    }
}