当前位置: 首页 > news >正文

【MacOS】RocketMQ 搭建Java客户端

【MacOS】RocketMQ 搭建Java客户端

文章目录

  • 【MacOS】RocketMQ 搭建Java客户端
    • 一、引入RocketMQ客户端依赖
      • 1.maven工程,在你的`pom.xml`中添加RocketMQ客户端依赖:
      • 2.gradle工程添加库
    • 二、创建生产者和消费者
      • 1.创建一个生产者
      • 消费者
        • 1.创建一个PullConsumer
        • 2.创建一个PushConsumer
    • 三、遇到的问题
      • 1.连接失败的问题
      • 2.主题没有找到

一、引入RocketMQ客户端依赖

1.maven工程,在你的pom.xml中添加RocketMQ客户端依赖:

<dependencies><!-- 添加RocketMQ客户端依赖 --><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>版本跟你下载的rocketmq版本一样</version></dependency>
</dependencies>

2.gradle工程添加库

compile 'org.apache.rocketmq:rocketmq-client:你的版本号'
  • 注意
    1. 客户端和服务端版本要一致,否则会发射管一些奇怪的问题
    2. 要到控制台创建Topic队列名称

二、创建生产者和消费者

1.创建一个生产者

mport com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;/*** @author pengxiaoping* @date 2024年10月18日 11:27*/
public class Producer {public static void main(String[] args) {Producer.producer();}public static void producer() {//创建DefaultMQProducer消息生产者对象DefaultMQProducer producer = new DefaultMQProducer("TestProducerGroup");//设置NameServer 多个节点间用分号分割producer.setNamesrvAddr("localhost:9876");try {//与NameServer建立长连接producer.start();//发送十条数据for (int i = 1; i <= 10; i++) {//1S中发送一次Thread.sleep(1000);JSONObject json = new JSONObject();json.put("orderId",i+1);json.put("desc","这是第"+i+1+"个订单");//数据正文String data = json.toJSONString();/*创建消息Message消息三个参数topic 代表消息主题,自定义自定义TopicOrder代表订单主题代表订单主题tags 代表标志,用于消费者接收数据时进行数据筛选。PAY_TAG代表支付相关信息body 代表消息内容*/Message message = new Message("TopicOrder", "PAY_TAG", data.getBytes());//发送消息,获取发送结果SendResult result = producer.send(message);//将发送结果对象打印在控制台System.out.println("消息已发送:MsgId:" + result.getMsgId() + ",发送状态:"+ result.getSendStatus());}}catch (Exception e){e.printStackTrace();}finally {try {producer.shutdown();} catch (Exception e) {}}}}

消费者

对于Consumer来说,他有两种基础的工作方式:pull和push。
区别:push:broker端来了消息以后主动将消息从broker端向consumer端推送。
pull:对于consumer来说主动往broker发一个请求,然后broker在通过请求响应给consumer一批消息。一般用push模式。

1.创建一个PullConsumer
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.common.message.MessageExt;import java.util.List;public class PullConsumer {public static volatile boolean running = true;public static void consumer() {//创建pull消费者对象DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("TestPullConsumerGroup");//设置NameServer节点litePullConsumer.setNamesrvAddr("localhost:9876");try {//订阅主题,与Push相同litePullConsumer.subscribe("TopicOrder", "*");//每次拉取数据条目数litePullConsumer.setPullBatchSize(10);//启动消费者litePullConsumer.start();while (running) {List<MessageExt> messageExts = litePullConsumer.poll();//批量数据处理for (MessageExt msg : messageExts) {System.out.println("消费者获取数据:" + msg.getMsgId() + "==>" + newString(msg.getBody()));}}}catch (Exception e){e.printStackTrace();}finally {litePullConsumer.shutdown();}}public static void main(String[] args) {PullConsumer.consumer();}
}
2.创建一个PushConsumer
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;import java.util.List;public class PushConsumer {public static void consumer() {//创建消费者对象DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TestConsumerGroup");try {//设置NameServer节点consumer.setNamesrvAddr("localhost:9876");/*订阅主题,consumer.subscribe包含两个参数:topic: 说明消费者从Broker订阅哪一个主题,这一项要与Provider保持一致。subExpression: 子表达式用于筛选tags。同一个主题下可以包含很多不同的tags,subExpression用于筛选符合条件的tags进行接收。例如:设置为*,则代表接收所有tags数据。例如:设置为PAY_TAG,则Broker中只有tags=PAY_TAG的消息会被接收,而其他的就会被排除在外。*/consumer.subscribe("TopicOrder", "*");//创建监听,当有新的消息监听程序会及时捕捉并加以处理。consumer.registerMessageListener(new MessageListenerConcurrently() {public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {//批量数据处理for (MessageExt msg : msgs) {System.out.println("消费者获取数据:" + msg.getMsgId() + "==>" + newString(msg.getBody()));}//返回数据已接收标识return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});//启动消费者,与Broker建立长连接,开始监听。consumer.start();} catch (Exception e) {e.printStackTrace();}}public static void main(String[] args) {PushConsumer.consumer();}
}

三、遇到的问题

1.连接失败的问题

Caused by: org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to null failedat org.apache.rocketmq.remoting.netty.NettyRemotingClient.invokeSync(NettyRemotingClient.java:572)at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:2050)at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:2041)at org.apache.rocketmq.client.impl.factory.MQClientInstance.updateTopicRouteInfoFromNameServer(MQClientInstance.java:782)... 6 more
  • 检查代码中设置的NameServer地址是否正确,跟配置文件中的NAME_ADDR地址一致。确保没有拼写错误、IP 地址或域名准确,以及端口号正确。
  • 检查 RocketMQ 的 NameServer 是否已经启动并且正在运行。查看 NameServer 的日志文件,确认没有错误或异常情况。如果 NameServer 没有启动,需要启动它。确保 NameServer 的配置正确,并且没有与其他服务冲突。

2.主题没有找到

org.apache.rocketmq.client.exception.MQClientException: No route info of this topic: TopicTest
See https://rocketmq.apache.org/docs/bestPractice/06FAQ for further details.at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:879)at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1564)at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:475)at org.apache.rocketmq.example.quickstart.Producer.main(Producer.java:78)
org.apache.rocketmq.client.exception.MQClientException: No route info of this topic: TopicTest
See https://rocketmq.apache.org/docs/bestPractice/06FAQ for further details.at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:879)at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1564)at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:475)at org.apache.rocketmq.example.quickstart.Producer.main(Producer.java:78)
  • 主题不存在或者未被创建:

    #在控制台创建主题 -n namesrv 的地址。-t 主题名。-c 指定所在集群
    sh bin/mqadmin updateTopic -n localhost:9876 -t TopicOrder -c DefaultCluster
    
    #出现这个创建成功
    create topic to 172.16.224.140:10911 success.
    TopicConfig [topicName=TopicOrder, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=false, attributes={}]
    

相关文章:

【MacOS】RocketMQ 搭建Java客户端

【MacOS】RocketMQ 搭建Java客户端 文章目录 【MacOS】RocketMQ 搭建Java客户端一、引入RocketMQ客户端依赖1.maven工程&#xff0c;在你的pom.xml中添加RocketMQ客户端依赖&#xff1a;2.gradle工程添加库 二、创建生产者和消费者1.创建一个生产者消费者1.创建一个PullConsume…...

前端学习---(5)js基础--3

ES 的全称是 ECMAScript&#xff0c;它是由 ECMA 国际标准化组织 制定的一套脚本语言的标准化规范。 ES6 的变量声明 let&#xff1a;定义变量 const&#xff1a;定义常量&#xff08;定义后&#xff0c;不可修改&#xff09; ES5中的 var 容易造成全局污染; ES6中的let可以在块…...

Spring Boot 3.3.4 升级导致 Logback 之前回滚策略配置不兼容问题解决

前言 在将 Spring Boot 项目升级至 3.3.4 版本后&#xff0c;遇到 Logback 配置的兼容性问题。本文将详细描述该问题的错误信息、原因分析&#xff0c;并提供调整日志回滚策略的解决方案。 错误描述 这是SpringBoot 3.3.3版本之前的回滚策略的配置 <!-- 日志记录器的滚动…...

如何开发属于自己的Hoobuy跨境独立站

以下是开发属于自己的类似 Pandabuy 或 Hoobuy 的跨境独立站的一般步骤&#xff1a; 市场调研与定位&#xff1a; 目标市场分析&#xff1a;确定您的独立站面向的海外目标市场&#xff0c;比如特定国家或地区。研究该市场的消费趋势、需求特点、竞争对手情况以及当地的法律法规…...

java智能物流管理系统源码(springboot)

项目简介 智能物流管理系统实现了以下功能&#xff1a; 智能物流管理系统的主要使用者分为管理员&#xff0c;顾客&#xff0c;员工&#xff0c;店主。功能有个人中心&#xff0c;顾客管理&#xff0c;员工管理&#xff0c;店主管理&#xff0c;门店信息管理&#xff0c;门店…...

全新语音图像数据集,以高质量训练数据加速提升模型性能

海天瑞声数据集上新&#xff1a;超60个国家地区口音英语语音识别数据集、多国口音西语语音识别数据集、印度多语种语音识别数据集、中文自然对话语音合成数据集、中文多音色语音合成数据集、多肤色高清人像数据集。海天瑞声高质量AI训练数据加速提升模型性能&#xff0c;让AI产…...

基于Springboot在线视频网站的设计与实现

基于Springboot视频网站的设计与实现 开发语言&#xff1a;Java 框架&#xff1a;springboot JDK版本&#xff1a;JDK1.8 服务器&#xff1a;tomcat7 数据库&#xff1a;mysql 5.7 数据库工具&#xff1a;Navicat11 开发软件&#xff1a;idea 源码获取&#xff1a;https://do…...

vue富文本使用editor

富文本【图片上传、缩放、拖动和不能复制只能根据点击图片上传到服务器】 <div id"editorId"><quill-editorref"myQuillEditor"v-model.trim"addForm.content":options"editorOption":disabled"isDisable"change&…...

Spring Boot植物健康系统:绿色科技的创新引擎

摘要 随着信息技术在管理上越来越深入而广泛的应用&#xff0c;管理信息系统的实施在技术上已逐步成熟。本文介绍了植物健康系统的开发全过程。通过分析植物健康系统管理的不足&#xff0c;创建了一个计算机管理植物健康系统的方案。文章介绍了植物健康系统的系统分析部分&…...

什么是域名?什么是泛域名?

域名 定义 域名是互联网上用于识别和定位网站或网络服务的名称。它是由一串用点分隔的字符组成&#xff0c;例如 “baidu.com”。就像是现实生活中建筑物的地址&#xff0c;方便用户在互联网的海量信息中找到特定的网站。 结构 域名从右到左依次为顶级域名&#xff08;TLD&…...

c#webapi远程调试方法asp.netcore

服务器安装 Visual Studio 2019 Remote Debugger 打开并运行 tools->Options 选择No Authorizaiton 确保IIS已经启动 打开本地项目->调试->添加到进程 参考&#xff1a; c#webapi远程调试方法asp.netcore - txwtech - 博客园...

XMLHttpRequest和FormData下载文件,ajax下载文件

1、前端请求只做下载功能 function downloadDatas3() {// 封装请求参数let formData new FormData();formData.append(page, 1);formData.append(rows, 10);// 创建xhr对象let xhr new XMLHttpRequest();xhr.open(POST, http://127.0.0.1:8080/getData);// xhr.setRequestHe…...

针对考研的C语言学习(2014二叉树大题代码实战)

题目描述 解析 1.递归思想遍历节点&#xff0c;若是叶子结点就累加计算的wpl&#xff0c;反之继续遍历 2.代码如下 //树 typedef struct trees {ElemType data;struct trees* lc;struct trees* rc; }treeNode, * Tree;3.算法设计 //deep路径长度也叫做深度&#xff0c;0开始 …...

webpack面试笔记(一)

一.webpack基础 1.模块化 什么是模块化? 模块化是把一个复杂的系统分解到多个模块以方便编码 为什么出现模块化 以前使用命名空间的方式来组织代码,比如jQuery,zepto, 它们有很多缺点: 命名空间冲突,两个库可能会使用同一个名称,例如zepto也被放在window.$下无法合理管理项目…...

雷池社区版有多个防护站点监听在同一个端口上,匹配顺序是怎么样的

如果域名处填写的分别为 IP 与域名&#xff0c;那么当使用进行 IP 请求时&#xff0c;则将会命中第一个配置的站点 以上图为例&#xff0c;如果用户使用 IP 访问&#xff0c;命中 example.com。 如果域名处填写的分别为域名与泛域名&#xff0c;除非准确命中域名&#xff0c;否…...

【小白学机器学习15】 概率论的世界观

目录 1 最近看的几本书和想说的 1.1 最近看的书 1.2 为什么写这个 2 概率论的观点看世界 2.1 上帝掷骰子&#xff0c;没有绝对的事情&#xff0c;所有事情都是概率决定的&#xff0c;都是相对的。 2.2 万物皆可能&#xff0c;无物是必然 2.3 什么是&#xff1a;可能性…...

合成数据用于大模型训练的3点理解

最近看国内对合成数据的研究讨论也变得多 ,而不单单是多模态,扩散模型这些偏视觉类的, 因此就合成数据写一下目前的情况。 2023年国外就有很多研究合成数据的论文, 包括Self-Consuming Generative Models Go MAD, Crowd Workers Widely Use Large Language Models for Text Pr…...

Safari 中 filter: blur() 高斯模糊引发的性能问题及解决方案

目录 引言问题背景&#xff1a;filter: blur() 引发的问题产生问题的原因分析解决方案&#xff1a;开启硬件加速实际应用示例性能优化建议常见的调试工具与分析方法 引言 在前端开发中&#xff0c;CSS滤镜&#xff08;如filter: blur()&#xff09;的广泛使用为页面带来了各种…...

浏览器实时更新esp32-c3 Supermini http server 数据

一利用此程序的思路就可以用浏览器显示esp32 采集的各种传感器的数据&#xff0c;也可以去控制各种传感器。省去编写针对各系统的app. 图片 1.浏览器每隔1秒更新一次数据 2.现在更新的是开机数据&#xff0c;运用此程序&#xff0c;可以实时显示各种传感器的实时数据 3.es…...

【亚马逊云】基于 Amazon EKS 搭建开源向量数据库 Milvus

文章目录 一、先决条件1.1 安装AWS CLI ✅1.2 安装 EKS 相关工具✅1.3 创建 Amazon S3 存储桶✅1.4 创建 Amazon MSK 实例✅ 二、创建EKS集群三、创建 ebs-sc StorageClass四、安装 AWS Load Balancer Controller五、部署 Milvus 数据库5.1 添加 Milvus Helm 仓库5.2 配置 S3 作…...

LBE-LEX系列工业语音播放器|预警播报器|喇叭蜂鸣器的上位机配置操作说明

LBE-LEX系列工业语音播放器|预警播报器|喇叭蜂鸣器专为工业环境精心打造&#xff0c;完美适配AGV和无人叉车。同时&#xff0c;集成以太网与语音合成技术&#xff0c;为各类高级系统&#xff08;如MES、调度系统、库位管理、立库等&#xff09;提供高效便捷的语音交互体验。 L…...

基于大模型的 UI 自动化系统

基于大模型的 UI 自动化系统 下面是一个完整的 Python 系统,利用大模型实现智能 UI 自动化,结合计算机视觉和自然语言处理技术,实现"看屏操作"的能力。 系统架构设计 #mermaid-svg-2gn2GRvh5WCP2ktF {font-family:"trebuchet ms",verdana,arial,sans-…...

css实现圆环展示百分比,根据值动态展示所占比例

代码如下 <view class""><view class"circle-chart"><view v-if"!!num" class"pie-item" :style"{background: conic-gradient(var(--one-color) 0%,#E9E6F1 ${num}%),}"></view><view v-else …...

微软PowerBI考试 PL300-选择 Power BI 模型框架【附练习数据】

微软PowerBI考试 PL300-选择 Power BI 模型框架 20 多年来&#xff0c;Microsoft 持续对企业商业智能 (BI) 进行大量投资。 Azure Analysis Services (AAS) 和 SQL Server Analysis Services (SSAS) 基于无数企业使用的成熟的 BI 数据建模技术。 同样的技术也是 Power BI 数据…...

Debian系统简介

目录 Debian系统介绍 Debian版本介绍 Debian软件源介绍 软件包管理工具dpkg dpkg核心指令详解 安装软件包 卸载软件包 查询软件包状态 验证软件包完整性 手动处理依赖关系 dpkg vs apt Debian系统介绍 Debian 和 Ubuntu 都是基于 Debian内核 的 Linux 发行版&#xff…...

(二)TensorRT-LLM | 模型导出(v0.20.0rc3)

0. 概述 上一节 对安装和使用有个基本介绍。根据这个 issue 的描述&#xff0c;后续 TensorRT-LLM 团队可能更专注于更新和维护 pytorch backend。但 tensorrt backend 作为先前一直开发的工作&#xff0c;其中包含了大量可以学习的地方。本文主要看看它导出模型的部分&#x…...

高等数学(下)题型笔记(八)空间解析几何与向量代数

目录 0 前言 1 向量的点乘 1.1 基本公式 1.2 例题 2 向量的叉乘 2.1 基础知识 2.2 例题 3 空间平面方程 3.1 基础知识 3.2 例题 4 空间直线方程 4.1 基础知识 4.2 例题 5 旋转曲面及其方程 5.1 基础知识 5.2 例题 6 空间曲面的法线与切平面 6.1 基础知识 6.2…...

GitHub 趋势日报 (2025年06月08日)

&#x1f4ca; 由 TrendForge 系统生成 | &#x1f310; https://trendforge.devlive.org/ &#x1f310; 本日报中的项目描述已自动翻译为中文 &#x1f4c8; 今日获星趋势图 今日获星趋势图 884 cognee 566 dify 414 HumanSystemOptimization 414 omni-tools 321 note-gen …...

是否存在路径(FIFOBB算法)

题目描述 一个具有 n 个顶点e条边的无向图&#xff0c;该图顶点的编号依次为0到n-1且不存在顶点与自身相连的边。请使用FIFOBB算法编写程序&#xff0c;确定是否存在从顶点 source到顶点 destination的路径。 输入 第一行两个整数&#xff0c;分别表示n 和 e 的值&#xff08;1…...

基于matlab策略迭代和值迭代法的动态规划

经典的基于策略迭代和值迭代法的动态规划matlab代码&#xff0c;实现机器人的最优运输 Dynamic-Programming-master/Environment.pdf , 104724 Dynamic-Programming-master/README.md , 506 Dynamic-Programming-master/generalizedPolicyIteration.m , 1970 Dynamic-Programm…...