一、rocketmq介绍

RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件，前身是MetaQ，是阿里参考Kafka特点研发的一个队列模型的消息中间件，后开源给apache基金会成为了apache的顶级开源项目，具有高性能、高可靠、高实时、分布式特点。

二、rocketmq环境搭建

采用docker-compose搭建，具体配置如下

version: '3'services:  # rocket mq name server  rmqnamesrv:    image: apache/rocketmq:4.9.6    restart: always    container_name: rocket-server      # environment:      #   JAVA_OPT_EXT: "-server -Xms64m -Xmx64m -Xmn64m"      # volumes:      # 映射本地目录权限一定要设置为 777 权限，否则启动不成功    # - ../volumes/data/rocket/server/logs:/home/rocketmq/logs    networks:      - rocketmq    ports:      - 9876:9876    command: sh mqnamesrv  # rocket mq broker  rmqbroker:    image: apache/rocketmq:4.9.6    restart: always    container_name: rocket-broker    volumes:      # 映射本地目录权限一定要设置为 777 权限，否则启动不成功      # - ../volumes/data/rocket/broker/logs:/home/rocketmq/logs      # - ../volumes/data/rocket/broker/store:/home/rocketmq/store      - ./config/broker.conf:/opt/rocketmq-4.9.6/conf/broker.conf    environment:      - NAMESRV_ADDR=rmqnamesrv:9876      # - JAVA_OPTS:=-Duser.home=/opt      - JAVA_OPT_EXT=-server -Xms64m -Xmx64m -Xmn64m    depends_on:      - rmqnamesrv    networks:      - rocketmq    ports:      - 10909:10909      - 10911:10911    command: sh mqbroker -c /opt/rocketmq-4.9.6/conf/broker.conf  # rocket console 这个可以不需要  rmqdashboard:    image: apacherocketmq/rocketmq-dashboard:1.0.0    restart: always    container_name: rocket-dashboard    environment:      - JAVA_OPTS=-Drocketmq.config.namesrvAddr=rmqnamesrv:9876 -Dserver.port=8180 -Drocketmq.config.isVIPChannel=false      # - JAVA_OPT_EXT=-Xms128m -Xmx128m -Xmn128m    depends_on:      - rmqnamesrv    networks:      - rocketmq    ports:      - 8180:8180
networks:  rocketmq:    driver: bridge  stack:    driver: bridge

运行docker-compose

docker-compose -f docker-compose-rocketmq.yml -p rocketmq up -d注：修改 xx/rocketmq/rocketmq_broker/conf/broker.conf中配置brokerIP1为宿主机IP

访问地址：http://ip地址:8180

三、代码工程

pom.xml

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">    <parent>        <artifactId>springboot-demo</artifactId>        <groupId>com.et</groupId>        <version>1.0-SNAPSHOT</version>    </parent>    <modelVersion>4.0.0</modelVersion>
    <artifactId>rocketmq</artifactId>
    <properties>        <maven.compiler.source>8</maven.compiler.source>        <maven.compiler.target>8</maven.compiler.target>    </properties>    <dependencies>        <dependency>            <groupId>org.apache.rocketmq</groupId>            <artifactId>rocketmq-spring-boot-starter</artifactId>            <version>2.1.1</version>        </dependency>        <dependency>            <groupId>org.apache.rocketmq</groupId>            <artifactId>rocketmq-client</artifactId>            <version>4.8.0</version>        </dependency>        <dependency>            <groupId>org.apache.rocketmq</groupId>            <artifactId>rocketmq-common</artifactId>            <version>4.8.0</version>        </dependency>        <dependency>            <groupId>junit</groupId>            <artifactId>junit</artifactId>            <scope>test</scope>        </dependency>        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-test</artifactId>            <scope>test</scope>        </dependency>        <dependency>            <groupId>org.springframework</groupId>            <artifactId>spring-test</artifactId>            <scope>test</scope>        </dependency>
    </dependencies>
</project>

entity

package com.et59.rocketmq.entity;
/** * @author liuhaihua * @version 1.0 * @ClassName Person * @Description todo * @date 2023年12月29日 9:52 */
public class Person {    private String name;    private int age;
    public String getName() {        return name;    }
    public void setName(String name) {        this.name = name;    }
    public int getAge() {        return age;    }
    public void setAge(int age) {        this.age = age;    }}

listener

package com.et59.rocketmq.listener;
import com.et59.rocketmq.entity.Person;import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;import org.apache.rocketmq.spring.core.RocketMQListener;import org.springframework.stereotype.Component;
/** * @Author: heyuhua * @Date: 2021/1/8 15:55 */@Component@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.groupName}", topic = "PERSON_ADD")public class PersonMqListener implements RocketMQListener<Person> {    @Override    public void onMessage(Person person) {        System.out.println("接收到消息，开始消费..name:" + person.getName() + ",age:" + person.getAge());    }}

util

package com.et59.rocketmq.util;
import org.apache.rocketmq.client.producer.SendCallback;import org.apache.rocketmq.client.producer.SendResult;import org.apache.rocketmq.spring.core.RocketMQTemplate;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.messaging.Message;import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;import javax.annotation.PreDestroy;
/** * RocketMq助手 * * @Author: heyuhua * @Date: 2020/1/8 10:03 */@Componentpublic class RocketMqHelper {
    /**     * 日志     */    private static final Logger LOG = LoggerFactory.getLogger(RocketMqHelper.class);
    /**     * rocketmq模板注入     */    @Autowired    private RocketMQTemplate rocketMQTemplate;
    @PostConstruct    public void init() {        LOG.info("---RocketMq助手初始化---");    }
    /**     * 发送异步消息     *     * @param topic   消息Topic     * @param message 消息实体     */    public void asyncSend(Enum topic, Message<?> message) {        asyncSend(topic.name(), message, getDefaultSendCallBack());    }

    /**     * 发送异步消息     *     * @param topic        消息Topic     * @param message      消息实体     * @param sendCallback 回调函数     */    public void asyncSend(Enum topic, Message<?> message, SendCallback sendCallback) {        asyncSend(topic.name(), message, sendCallback);    }
    /**     * 发送异步消息     *     * @param topic   消息Topic     * @param message 消息实体     */    public void asyncSend(String topic, Message<?> message) {        rocketMQTemplate.asyncSend(topic, message, getDefaultSendCallBack());    }
    /**     * 发送异步消息     *     * @param topic        消息Topic     * @param message      消息实体     * @param sendCallback 回调函数     */    public void asyncSend(String topic, Message<?> message, SendCallback sendCallback) {        rocketMQTemplate.asyncSend(topic, message, sendCallback);    }
    /**     * 发送异步消息     *     * @param topic        消息Topic     * @param message      消息实体     * @param sendCallback 回调函数     * @param timeout      超时时间     */    public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout) {        rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout);    }
    /**     * 发送异步消息     *     * @param topic        消息Topic     * @param message      消息实体     * @param sendCallback 回调函数     * @param timeout      超时时间     * @param delayLevel   延迟消息的级别     */    public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout, int delayLevel) {        rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout, delayLevel);    }
    /**     * 发送顺序消息     *     * @param message     * @param topic     * @param hashKey     */    public void syncSendOrderly(Enum topic, Message<?> message, String hashKey) {        syncSendOrderly(topic.name(), message, hashKey);    }

    /**     * 发送顺序消息     *     * @param message     * @param topic     * @param hashKey     */    public void syncSendOrderly(String topic, Message<?> message, String hashKey) {        LOG.info("发送顺序消息，topic:" + topic + ",hashKey:" + hashKey);        rocketMQTemplate.syncSendOrderly(topic, message, hashKey);    }
    /**     * 发送顺序消息     *     * @param message     * @param topic     * @param hashKey     * @param timeout     */    public void syncSendOrderly(String topic, Message<?> message, String hashKey, long timeout) {        LOG.info("发送顺序消息，topic:" + topic + ",hashKey:" + hashKey + ",timeout:" + timeout);        rocketMQTemplate.syncSendOrderly(topic, message, hashKey, timeout);    }
    /**     * 默认CallBack函数     *     * @return     */    private SendCallback getDefaultSendCallBack() {        return new SendCallback() {            @Override            public void onSuccess(SendResult sendResult) {                LOG.info("---发送MQ成功---");            }
            @Override            public void onException(Throwable throwable) {                throwable.printStackTrace();                LOG.error("---发送MQ失败---"+throwable.getMessage(), throwable.getMessage());            }        };    }

    @PreDestroy    public void destroy() {        LOG.info("---RocketMq助手注销---");    }
}

配置文件

server:  port: 8088#rocketmq配置rocketmq:  name-server: 10.11.68.77:9876  # 生产者配置  producer:    isOnOff: on    # 发送同一类消息的设置为同一个group，保证唯一    group: hyh-rocketmq-group    groupName: hyh-rocketmq-group    # 服务地址    namesrvAddr: 10.11.68.77:9876    # 消息最大长度 默认1024*4(4M)    maxMessageSize: 4096    # 发送消息超时时间,默认3000    sendMsgTimeout: 3000    # 发送消息失败重试次数，默认2    retryTimesWhenSendFailed: 2

启动类

package com.et59.rocketmq;
import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplicationpublic class DemoApplication {
   public static void main(String[] args) {      SpringApplication.run(DemoApplication.class, args);   }}

code repository

https://github.com/Harries/springboot-demo

四、测试

package com.et59.rocketmq;

import com.et59.rocketmq.entity.Person;import com.et59.rocketmq.util.RocketMqHelper;import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.messaging.support.MessageBuilder;import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)@SpringBootTestpublic class MQTests {   @Autowired   private RocketMqHelper rocketMqHelper;
   @Test   public void testProducter() throws InterruptedException {      for(int i=0;i<1000000;i++){         Person person = new Person();         person.setName("heyuhua");         person.setAge(25);         rocketMqHelper.asyncSend("PERSON_ADD", MessageBuilder.withPayload(person).build());         Thread.sleep(1000);      }   }
}

运行结果

2024-02-02 15:20:27.101 INFO 11188 --- [ublicExecutor_3] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---2024-02-02 15:20:28.116 INFO 11188 --- [ublicExecutor_4] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---2024-02-02 15:20:29.130 INFO 11188 --- [ublicExecutor_5] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---2024-02-02 15:20:30.131 INFO 11188 --- [ublicExecutor_7] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---2024-02-02 15:20:31.142 INFO 11188 --- [ublicExecutor_8] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---2024-02-02 15:20:32.156 INFO 11188 --- [ublicExecutor_9] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---2024-02-02 15:20:33.167 INFO 11188 --- [blicExecutor_10] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---接收到消息，开始消费..name:heyuhua,age:25接收到消息，开始消费..name:heyuhua,age:25接收到消息，开始消费..name:heyuhua,age:25接收到消息，开始消费..name:heyuhua,age:25接收到消息，开始消费..name:heyuhua,age:252024-02-02 15:20:34.182 INFO 11188 --- [ublicExecutor_3] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---接收到消息，开始消费..name:heyuhua,age:25接收到消息，开始消费..name:heyuhua,age:25接收到消息，开始消费..name:heyuhua,age:25接收到消息，开始消费..name:heyuhua,age:25接收到消息，开始消费..name:heyuhua,age:25接收到消息，开始消费..name:heyuhua,age:25

五、引用

https://rocketmq.apache.org/
http://www.liuhaihua.cn/archives/710205.html

ITPub博客

Springboot集成rocketmq快速入门demo

一、rocketmq介绍

二、rocketmq环境搭建

三、代码工程

pom.xml

entity

listener

util

配置文件

启动类

code repository

四、测试

五、引用

最新文章