Springboot 整合 RocketMQ 收发消息的配置过程

2024-04-02 19:04:59 593人浏览八月长安

Python 官方文档：入门教程 => 点击学习

摘要

SpringBoot 整合 RocketMQ 收发消息创建springboot项目 pom.xml添加rocketMQ-spring-boot-starter依赖。 <d

SpringBoot 整合 RocketMQ 收发消息

创建springboot项目

pom.xml添加rocketMQ-spring-boot-starter依赖。


<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.0</version>
</dependency>

yml 配置

application.yml


rocketmq:
  name-server: 192.168.64.141:9876

application-demo1.yml

使用 demo1 profile 指定生产者组组名


rocketmq:
  producer:
    group: producer-demo1

application-demo2.yml

使用 demo2 profile 指定生产者组组名


rocketmq:
  producer:
    group: producer-demo2

测试

demo 1

发送普通消息
发送 Spring 的通用 Message 对象
发送异步消息
发送顺序消息

生产者


package cn.tedu.demo2.m1;
import org.apache.rocketmq.client.producer.SendCallback;
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.stereotype.Component;
@Component
public class Producer {
    @Autowired
    private RocketMQTemplate t ;
    public  void send(){
        //发送同步消息
        t.convertAndSend("Topic1:TagA", "Hello world! ");
        //发送spring的Message
        Message<String> message = MessageBuilder.withPayload("Hello Spring message! ").build();
        t.send("Topic1:TagA",message);
        //发送异步消息
        t.asyncSend("Topic1:TagA", "hello world asyn", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送成功");
            }
            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送失败");
            }
        });
        //发送顺序消息
        t.syncSendOrderly("Topic1", "98456237,创建", "98456237");
        t.syncSendOrderly("Topic1", "98456237,支付", "98456237");
        t.syncSendOrderly("Topic1", "98456237,完成", "98456237");
    }
}

消费者


package cn.tedu.demo2.m1;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Component
@RocketMQMessageListener(topic = "Topic1",consumerGroup = "consumer-demo1")
public class Consumer  implements RocketMQListener<String> {
    @Override
    public void onMessage(String s) {
        System.out.println("收到"+s);
    }
}

主类


package cn.tedu.demo2.m1;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Main {
    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
    }
}

测试类

需要放在 test 文件夹

激活 demo1 profile @ActiveProfiles("demo1")


package cn.tedu.demo2.m1;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.ActiveProfiles;
@SpringBootTest
@ActiveProfiles("demo1")
public class Test1 {
    @Autowired
    private  Producer producer;
    @Test
    public void test1(){
        producer.send();
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

demo 2

发送事务消息

生产者


package cn.tedu.demo2.m2;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
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.stereotype.Component;

@Component

public class Producer {

    @Autowired
    private RocketMQTemplate t;

    public void send(){
        Message<String> message = MessageBuilder.withPayload("Hello world").build();
        //一旦发送消息，则执行监听器
        t.sendMessageInTransaction("Topic2",message,null);
    }
    @RocketMQTransactionListener
    class Lis implements RocketMQLocalTransactionListener {
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
            System.out.println("执行本地事务");
            return RocketMQLocalTransactionState.UNKNOWN;
        }

        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
            System.out.println("执行事务回查");
            return RocketMQLocalTransactionState.COMMIT;
        }
    }

}

消费者


package cn.tedu.demo2.m2;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Component
@RocketMQMessageListener(topic = "Topic2",consumerGroup = "consumer-demo2")
public class Consumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String s) {
        System.out.println("收到"+s);
    }
}

主类


package cn.tedu.demo2.m2;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Main {
    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
    }
}

测试类


package cn.tedu.demo2.m2;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.ActiveProfiles;
@SpringBootTest
@ActiveProfiles("demo2")
public class Test2 {
    @Autowired
    private  Producer producer;
    @Test
    public void  test1(){
        producer.send();
        //为了能够收到消费者消费的数据，这里通过休眠模拟等待时间
        try {
            Thread.sleep(30000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

到此这篇关于Springboot 整合 RocketMQ 收发消息的文章就介绍到这了,更多相关Springboot 整合 RocketMQ 收发消息内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

您可能感兴趣的文档:

点击免费下载>>软考高级考试备考技巧/历年真题/备考精华资料

--结束END--

本文标题: Springboot 整合 RocketMQ 收发消息的配置过程

本文链接: https://www.lsjlt.com/news/161168.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

本篇文章演示代码以及资料文档资料下载

下载Word文档到电脑，方便收藏和打印～

下载Word文档

去做题

猜你喜欢

Springboot 整合 RocketMQ 收发消息的配置过程

Springboot 整合 RocketMQ 收发消息创建springboot项目 pom.xml添加rocketmq-spring-boot-starter依赖。 <d...

99+

2022-11-12
Springboot如何整合RocketMQ收发消息

这篇文章将为大家详细讲解有关Springboot如何整合RocketMQ收发消息，小编觉得挺实用的，因此分享给大家做个参考，希望大家阅读完这篇文章后可以有所收获。Springboot 整合 RocketMQ 收发消息创建springboot...

99+

2023-06-22
SpringBoot整合RocketMQ实现消息发送和接收的详细步骤

我们使用主流的SpringBoot框架整合RocketMQ来讲解，使用方便快捷；最终项目结构如下：具体步骤如下：第一步：新建SpringBoot项目rocketmq-test...

99+

2022-11-12
SpringBoot整合RocketMQ的详细过程

目录1. SpringBoot整合RocketMQ2 使用RocketMQ会遇到的问题2.1 WARN No appenders could be found for logger2...

99+

2023-05-15

SpringBoot整合RocketMQ SpringBoot RocketMQ使用
Springboot详解RocketMQ实现消息发送与接收流程

springboot+rockermq 实现简单的消息发送与接收普通消息的发送方式有3种：单向发送、同步发送和异步发送。下面来介绍下 springboot+rockermq 整合...

99+

2022-11-13
springboot整合seata的配置过程

前言：小编引入的图片和文字描述都是来自于尚硅谷的视频讲解，在此感谢尚硅谷的老师,同时也结合 seata文档官方文档进行整合项目地址（gitee）: https://gitee.co...

99+

2022-11-12
Springboot整合ActiveMQ实现消息队列的过程浅析

目录pom中导入坐标书写yml配置业务层代码监听器代码业务层代码确保你启动了自己电脑的activemq。 pom中导入坐标 <dependency> ...

99+

2023-02-10

Springboot整合ActiveMQ Springboot ActiveMQ消息队列
Springboot详细讲解RocketMQ实现顺序消息的发送与消费流程

目录一、创建Springboot项目添加rockermq依赖二、配置rocketmq三、新建一个controller来做消息发送四、创建消费端监听消息消费消息五、启动服务测试顺序消息...

99+

2022-11-13
Android如何通过接收UDP消息改写程序配置

今天就跟大家聊聊有关Android如何通过接收UDP消息改写程序配置，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。Android程序打包成APK发布后，如果有参数需要根据不同应用环境...

99+

2023-06-05
SpringBoot整合Redis实现消息发布与订阅的示例代码

当我们在多个集群应用中使用到本地缓存时，在数据库数据得到更新后，为保持各个副本当前被修改的数据与数据库数据保持同步，在数据被操作后向其他集群应用发出被更新数据的通知，使其删除;下次当...

99+

2022-11-13

SpringBoot Redis 消息发布 SpringBoot Redis 消息订阅 SpringBoot Redis 消息发布订阅 SpringBoot Redis
Idea开发工具之SpringBoot整合JSP的过程

SpringBoot体系内推荐使用Thymeleaf作为前端页面模板。jsp还得自己整合一下。 1.项目结构对比以前的项目结构，main目录下多了webapp目录，用来存放jsp...

99+

2022-11-12
springboot整合mybatis的超详细过程(配置模式+注解模式)

目录一.简单介绍二具体配置2.1.配置相关的依赖.2.2 写.mapper、controller、service2.2.1mapper文件2.2.2service文件2.2.2con...

99+

2022-11-13
springboot整合netty-mqtt-client实现Mqtt消息的订阅和发布示例

目录1.添加依赖2.源码3.运行测试1.添加依赖 <dependency> <groupId>org.jetlinks</groupId>...

99+

2022-11-13
SpringBoot整合WebSocket实现后端向前端发送消息的实例代码

一、什么是 websocket 接口使用 websocket 建立长连接，服务端和客户端可以互相通信，服务端只要有数据更新，就可以主动推给客户端。 WebSocket 使得客户端和...

99+

2023-03-06

springboot整合websocket推送 spring boot整合websocket springboot主动推送前端
SpringBoot整合Javamail实现邮件发送的详细过程

目录序言简单邮件多组件邮件（带附件，图片，超链接的邮件）发送带超链接的邮件发送带图片的邮件总结序言发邮件对于咱们的Java来说是一个基本的功能，因此这里SpringBoot整合...

99+

2022-11-13

springboot整合javamail springboot 邮件发送
Spring配置文件的拆分和整合过程分析

目录一、Spring配置文件拆分：二、Spring配置文件整合：一、Spring配置文件拆分：在实际应用里，随着应用规模的增加，系统中 Bean 数量也大量增加，导致配置文件非常庞...

99+

2022-11-13