springboot整合rocketmq

发布于: 2022-08-08 更新于: 2022-09-29 分类于: Java 阅读次数:

一：pom引入依赖

<!-- rocketmq start -->
<dependency>
	<groupId>org.apache.rocketmq</groupId>
	<artifactId>rocketmq-spring-boot-starter</artifactId>
	<version>2.2.2</version>
</dependency>
<!-- rocketmq end -->

二：yml文件

# rocketmq 消息队列
rocketmq:
  name-server: 127.0.01:9876 #server地址
  # 生产者
  producer:
    #生产者组名，规定在一个应用里面必须唯一
    group: jtools_group
    topic: jtools_topic
    tag: jtools_tag
    #消息发送的超时时间，默认为3000ms
    send-message-timeout: 3000
    #消息达到4096字节的时候，消息就会被压缩。默认4096
    compress-message-body-threshold: 4096
    #最大的消息限制 默认为128K
    max-message-size: 4194304
    #同步消息发送失败重试次数
    retry-times-when-send-failed: 3
    #在内部发送失败时是否重试其他代理,这个参数在有多个broker才生效。
    retry-next-server: true
    #异步消息发送失败重试的次数
    retry-times-when-send-async-failed: 3
  # 消费者
  consumer:
    topic: ${rocketmq.producer.topic}
    group: ${rocketmq.producer.group}
    tag: ${rocketmq.producer.tag}

1 2	我这里整合了生产者和消费者，并且消费者的topic要和生产的一致，我这里全部用的自定义的applicationName

三：生产者

Controller测试

package com.tohours.jtools.controller;

import java.util.Date;

import org.apache.rocketmq.client.producer.SendResult;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.tohours.jtools.annotation.WebLog;
import com.tohours.jtools.support.TohoursResponse;
import com.tohours.jtools.support.TohoursResult;
import com.tohours.jtools.utils.RocketMqUtils;
import com.tohours.jtools.utils.TohoursUtils;

/**
 * @desc roketmq
 */
@RestController
@RequestMapping("/api/mq")
public class RocketMqController {

    /**
     * @desc 发送消息
     * @return
     */
    @GetMapping("send-msg")
	@WebLog
    public TohoursResult<Object> sendMsg() {
        String time = TohoursUtils.date2string(new Date(),"yyyy-MM-dd HH:mm:ss");
        //发送普通消息
        RocketMqUtils.sendMessage("普通消息 当前时间：" + time);
        return TohoursResponse.success();
    }
    
    /**
     * @desc 发送消息
     * @return
     */
    @GetMapping("send-msg2")
	@WebLog
    public TohoursResult<Object> sendMsg2() {
        String time = TohoursUtils.date2string(new Date(),"yyyy-MM-dd HH:mm:ss");
        SendResult result = RocketMqUtils.syncSendMessage("同步消息 当前时间：" + time);
        return TohoursResponse.success(result);
    }
    
    
    /**
     * @desc 发送消息
     * @return
     */
    @GetMapping("send-msg3")
	@WebLog
    public TohoursResult<Object> sendMsg3() {
        String time = TohoursUtils.date2string(new Date(),"yyyy-MM-dd HH:mm:ss");
        SendResult result = RocketMqUtils.syncSendMessage("test-tag","当前时间：" + time);
        return TohoursResponse.success(result);
    }


}

MQ.java

package com.tohours.jtools.constants;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import lombok.Data;

/**
 * @desc MQ的相关信息
 */
@Component
public @Data class MQ {

	// server地址
	@Value("${rocketmq.name-server}")
	private String nameServer;

	// 生产者group
	@Value("${rocketmq.producer.group}")
	private String producerGroup;

	// 生产者topic
	@Value("${rocketmq.producer.topic}")
	private String producerTopic;
	
	// 生产者tag
	@Value("${rocketmq.producer.tag}")
	private String producerTag;

	// 消费者group
	@Value("${rocketmq.consumer.group}")
	private String consumerGroup;

	// 消费者topic
	@Value("${rocketmq.consumer.topic}")
	private String consumerTopic;

	// 消费者tag
	@Value("${rocketmq.consumer.tag}")
	private String consumerTag;

}

RocketMQUtils.java

package com.tohours.jtools.utils;

import java.util.List;

import javax.annotation.PostConstruct;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.tohours.jtools.constants.MQ;

import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
public class RocketMqUtils {

	@Autowired
	private MQ cmq;
	@Autowired
	private RocketMQTemplate crocketMQTemplate;

	private static MQ mq;
	private static RocketMQTemplate rocketMQTemplate;

	@PostConstruct
	public void init() {
		mq = cmq;
		rocketMQTemplate = crocketMQTemplate;
	}
	
	/**
	 * @desc 构建目的地
	 * @param tag 标签
	 * @return
	 */
	public static String buildDestination() {
		return buildDestination(mq.getProducerTopic(), mq.getProducerTag());
	}

	/**
	 * @desc 构建目的地
	 * @param tag 标签
	 * @return
	 */
	public static String buildDestinationWithTag(String tag) {
		return buildDestination(mq.getProducerTopic(), tag);
	}
	
	/**
	 * @desc 构建目的地
	 * @param topic 主题
	 * @return
	 */
	public static String buildDestinationWithTopic(String topic) {
		return buildDestination(topic, mq.getProducerTag());
	}

	/**
	 * @desc 构建目的地
	 * @param topic 主题
	 * @param tag   标签
	 * @return
	 */
	public static String buildDestination(String topic, String tag) {
		log.info("topic:" + topic);
		log.info("tag:" + tag);
		return topic + ":" + tag;
	}

	/**
	 * @desc 发送普通消息
	 * @param message 消息体
	 */
	public static void sendMessage(Object message) {
		sendMessage(mq.getProducerTopic(), message);
	}
	
	/**
	 * @desc 发送普通消息
	 * @param topic   主题
	 * @param message 消息体
	 */
	public static void sendMessage(String topic, Object message) {
		rocketMQTemplate.convertAndSend(buildDestinationWithTopic(topic), message);
		log.info("普通消息发送完成：message = {}", message);
	}

	/**
	 * @desc 发送普通消息
	 * @param topic   主题
	 * @param tag   标签
	 * @param message 消息体
	 */
	public static void sendMessage(String topic,String tag, Object message) {
		rocketMQTemplate.convertAndSend(buildDestination(topic,tag), message);
		log.info("普通消息发送完成：message = {}", message);
	}

	/**
	 * @desc 发送同步消息
	 * @param message 消息体
	 */
	public static SendResult syncSendMessage(Object message) {
		return syncSendMessage(mq.getProducerTopic(),mq.getProducerTag(), message);
	}
	
	/**
	 * @desc 发送同步消息
	 * @param tag 标签
	 * @param message 消息体
	 */
	public static SendResult syncSendMessage(String tag, Object message) {
		SendResult sendResult = rocketMQTemplate.syncSend(buildDestinationWithTag(tag),message);
		log.info("同步发送消息完成：message = {}, sendResult = {}", message, sendResult);
		return sendResult;
	}

	/**
	 * @desc 发送同步消息
	 * @param topic   主题
	 * @param tag 标签
	 * @param message 消息体
	 */
	public static SendResult syncSendMessage(String topic,String tag, Object message) {
		SendResult sendResult = rocketMQTemplate.syncSend(buildDestination(topic,tag),message);
		log.info("同步发送消息完成：message = {}, sendResult = {}", message, sendResult);
		return sendResult;
	}

	/**
	 * @desc 发送异步消息
	 * @param message 消息体
	 */
	public static void asyncSendMessage(Object message) {
		asyncSendMessage(mq.getProducerTopic(),mq.getProducerTag(), message);
	}
	
	/**
	 * @desc 发送异步消息
	 * @pram tag 标签
	 * @param message 消息体
	 */
	public static void asyncSendMessage(String tag,Object message) {
		asyncSendMessage(mq.getProducerTopic(),tag, message);
	}

	/**
	 * @desc 发送异步消息
	 * @param topic   主题
	 * @param tag 标签
	 * @param message 消息体
	 */
	public static void asyncSendMessage(String topic,String tag, Object message) {
		log.info("topic:" + topic);
		rocketMQTemplate.asyncSend(buildDestination(topic,tag), message, new SendCallback() {
			@Override
			public void onSuccess(SendResult sendResult) {
				log.info("异步消息发送成功，message = {}, SendStatus = {}", message, sendResult.getSendStatus());
			}

			@Override
			public void onException(Throwable e) {
				log.info("异步消息发送异常，exception = {}", e.getMessage());
			}
		});
	}

	/**
	 * @desc 发送单向消息
	 * @param message 消息体
	 */
	public static void sendOneWayMessage(Object message) {
		sendOneWayMessage(mq.getProducerTopic(), message);
	}
	
	/**
	 * @desc 发送单向消息
	 * @param tag 标签
	 * @param message 消息体
	 */
	public static void sendOneWayMessage(String tag,Object message) {
		sendOneWayMessage(mq.getProducerTopic(),mq.getProducerTag(), message);
	}

	/**
	 * @desc 发送单向消息
	 * @param topic   主题
	 * @param tag 标签
	 * @param message 消息体
	 */
	public static void sendOneWayMessage(String topic,String tag, Object message) {
		rocketMQTemplate.sendOneWay(buildDestinationWithTag(tag), message);
		log.info("单向发送消息完成：message = {}", message);
	}

	/**
	 * @desc 同步发送批量消息
	 * @param messageList 消息集合
	 * @param timeout     超时时间（毫秒）
	 */
	public static SendResult syncSendMessages(List<Message> messageList, long timeout) {
		return syncSendMessages(mq.getProducerTopic(),mq.getProducerTag(), messageList, timeout);
	}
	
	/**
	 * @desc 同步发送批量消息
	 * @param tag 标签
	 * @param messageList 消息集合
	 * @param timeout     超时时间（毫秒）
	 */
	public static SendResult syncSendMessages(String tag,List<Message> messageList, long timeout) {
		return syncSendMessages(mq.getProducerTopic(),tag, messageList, timeout);
	}

	/**
	 * @desc 同步发送批量消息
	 * @param topic       主题
	 * @param tag       标签
	 * @param messageList 消息集合
	 * @param timeout     超时时间（毫秒）
	 */
	public static SendResult syncSendMessages(String topic,String tag, List<Message> messageList, long timeout) {
		log.info("topic:" + topic);
		SendResult result = rocketMQTemplate.syncSend(buildDestination(topic,tag), messageList, timeout);
		log.info("同步发送批量消息完成：message = {}，result = {}", JSON.toJSONString(messageList), JSON.toJSONString(result));
		return result;
	}

	/**
	 * @desc 异步发送延时消息
	 * @param message    消息对象
	 * @param timeout    超时时间
	 * @param delayLevel 延时等级 默认18个等级，0表示不延时，messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
	 * 					level == 0，消息为非延迟消息
						1<=level<=maxLevel，消息延迟特定时间，例如level==1，延迟1s
						level > maxLevel，则level== maxLevel，例如level==20，延迟2h
	 */
	public static void asyncSendDelay(Object message, long timeout, int delayLevel) {
		asyncSendDelay(mq.getProducerTag(), message, timeout, delayLevel);
	}
	
	/**
	 * @desc 异步发送延时消息
	 * @param tag    主题
	 * @param message    消息对象
	 * @param timeout    超时时间
	 * @param delayLevel 延时等级 默认18个等级，0表示不延时，messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
	 * 					level == 0，消息为非延迟消息
						1<=level<=maxLevel，消息延迟特定时间，例如level==1，延迟1s
						level > maxLevel，则level== maxLevel，例如level==20，延迟2h
	 */
	public static void asyncSendDelay(String tag,Object message, long timeout, int delayLevel) {
		asyncSendDelay(mq.getProducerTopic(),mq.getProducerTag(), message, timeout, delayLevel);
	}

	/**
	 * @desc 异步发送延时消息
	 * @param topic      主题
	 * @param tag 标签
	 * @param message    消息对象
	 * @param timeout    超时时间
	 * @param delayLevel 延时等级 默认18个等级，0表示不延时，messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
	 * 					level == 0，消息为非延迟消息
						1<=level<=maxLevel，消息延迟特定时间，例如level==1，延迟1s
						level > maxLevel，则level== maxLevel，例如level==20，延迟2h
	 */
	public static void asyncSendDelay(String topic,String tag, Object message, long timeout, int delayLevel) {
		rocketMQTemplate.asyncSend(buildDestination(topic,tag), MessageBuilder.withPayload(message).build(), new SendCallback() {
			@Override
			public void onSuccess(SendResult sendResult) {
				log.info("异步发送延时消息成功，message = {}", message);
			}

			@Override
			public void onException(Throwable throwable) {
				log.error("异步发送延时消息发生异常，exception = {}", throwable.getMessage());
			}
		}, timeout, delayLevel);
		log.info("已异步发送延时消息 message = {}", message);
	}

}

四：消费者

package com.tohours.jtools.controller;

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

import lombok.extern.slf4j.Slf4j;


/**
 * @desc 消费者监听
 */
@SuppressWarnings("rawtypes")
@Component
@RocketMQMessageListener(topic = "${rocketmq.consumer.topic}" ,consumerGroup = "${rocketmq.consumer.group}") //topic要与生产者的topic一致
@Slf4j
public class RocketMqConsumer implements RocketMQListener {

    @Override
    public void onMessage(Object message) {
        log.info("message:" + message);
    }
}

*************感谢您的阅读*************