springBoot整合RocketMQ及坑的示例代码

版本:

  • JDK:1.8
  • springBoot:1.5.10
  • rocketMQ:4.2.0

pom 配置:

<parent>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-parent</artifactId>
 <version>1.5.10.RELEASE</version>
</parent>
<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-client</artifactId>
  <version>4.2.0</version>
</dependency>

application.properties  配置:

# 消费者的组名
apache.rocketmq.consumer.PushConsumer=PushConsumer
# 生产者的组名
apache.rocketmq.producer.producerGroup=Producer
# NameServer地址
apache.rocketmq.namesrvAddr=localhost:9876

java代码:

生产者

package test.config.rocketmq;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;
import javax.annotation.PostConstruct;

@Component
public class RocketMQClient {
  /**
   * 生产者的组名
   */
  @Value("${apache.rocketmq.producer.producerGroup}")
  private String producerGroup;

  /**
   * NameServer 地址
   */
  @Value("${apache.rocketmq.namesrvAddr}")
  private String namesrvAddr;

  @PostConstruct
  public void defaultMQProducer() {
    //生产者的组名
    DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
    //指定NameServer地址,多个地址以 ; 隔开
    producer.setNamesrvAddr(namesrvAddr);
    producer.setVipChannelEnabled(false);
    try {
      /**
       * Producer对象在使用之前必须要调用start初始化,初始化一次即可
       * 注意:切记不可以在每次发送消息时,都调用start方法
       */
      producer.start();

      //创建一个消息实例,包含 topic、tag 和 消息体
      //如下:topic 为 "TopicTest",tag 为 "push"
      Message message = new Message("TopicTest", "push", "发送消息----zhisheng-----".getBytes(RemotingHelper.DEFAULT_CHARSET));

      StopWatch stop = new StopWatch();
      stop.start();

      for (int i = 0; i < 1; i++) {
        SendResult result = producer.send(message);
        System.out.println("发送响应:MsgId:" + result.getMsgId() + ",发送状态:" + result.getSendStatus());
      }
      stop.stop();
      System.out.println("----------------发送一万条消息耗时:" + stop.getTotalTimeMillis());
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      producer.shutdown();
    }
  }
}

消费者:

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Component
public class RocketMQServer {
  /**
   * 消费者的组名
   */
  @Value("${apache.rocketmq.consumer.PushConsumer}")
  private String consumerGroup;

  /**
   * NameServer 地址
   */
  @Value("${apache.rocketmq.namesrvAddr}")
  private String namesrvAddr;

  @PostConstruct
  public void defaultMQPushConsumer() {
    //消费者的组名
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);

    //指定NameServer地址,多个地址以 ; 隔开
    consumer.setNamesrvAddr(namesrvAddr);
    consumer.setVipChannelEnabled(false);
    try {
      //订阅PushTopic下Tag为push的消息
      consumer.subscribe("TopicTest", "push");

      //设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
      //如果非第一次启动,那么按照上次消费的位置继续消费
      consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
      consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
        try {
          for (MessageExt messageExt : list) {

            System.out.println("messageExt: " + messageExt);//输出消息内容

            String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);

            System.out.println("消费响应:msgId : " + messageExt.getMsgId() + ", msgBody : " + messageBody);//输出消息内容
          }
        } catch (Exception e) {
          e.printStackTrace();
          return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
      });
      consumer.start();
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
}

掉坑总结:

1.rocketMQ启动时,命令不是  mqbroker -n 127.0.0.1:9876

正确应该是:mqbroker -n 127.0.0.1:9876 butiautoCreateTopicEnable=true

否则会抛出:No route info of this topic, TopicTest

2.客户端连接时抛出异常

org.apache.rocketmq.client.exception.MQClientException:

Send [3] times, still failed, cost [3180]ms, Topic: TopicTest, BrokersSent: \

[WIN-93CGO0S5G25, WIN-93CGO0S5G25, WIN-93CGO0S5G25]

解决方式两种

1.producer.setVipChannelEnabled(false); 生产者和消费者添加这行代买。

2.降rocketmq版本,降成3.2.6

关于spring.rocketmq.name-server的坑

看下图:

注意:

如果你是SpringBoot2.0+的框架,或者是JDK10。

你需要将你自己的项目配置文件中的,spring.rocketmq.name-server改成

spring.rocketmq.nameServer。注意是nameServer。

不然就会报各种稀奇古怪的bug。

关于启动报内存不足的错

在安装启动Name Server和Broker的时候,一定要修改配置文件,不然内存会爆炸。

Native memory allocation (mmap) failed to map 8589934592 bytes for committing reserved memory

将下面的配置文件根据你的需要改

我这里以前默认是Xms4g,都是g,我修改到m就行了。

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

(0)

相关推荐

  • rocketmq消费负载均衡--push消费详解

    前言 本文介绍了DefaultMQPushConsumerImpl消费者,客户端负载均衡相关知识点.本文从DefaultMQPushConsumerImpl启动过程到实现负载均衡,从源代码一步一步分析,共分为6个部分进行介绍,其中第6个部分 rebalanceByTopic 为负载均衡的核心逻辑模块,具体过程运用了图文进行阐述. 介绍之前首先抛出几个问题: 1. 要做负载均衡,首先要解决的一个问题是什么? 2. 负载均衡是Client端处理还是Broker端处理? 个人理解: 1. 要做负载均衡

  • java rocketmq--消息的产生(普通消息)

    前言 与消息发送紧密相关的几行代码: 1. DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName"); 2. producer.start(); 3. Message msg = new Message(...) 4. SendResult sendResult = producer.send(msg); 5. producer.shutdown(); 那这几行代码执行时,背后都做了什么? 一. 首先

  • Window搭建部署RocketMQ步骤详解

    序 以前简单用过ActiveMQ但是公司项目上使用的是RocketMQ,所以准备多花点时间在这上面,搞懂项目的配置使用. 看了很多资料,先说说我自己对RocketMQ的简单理解.不管是我们写的消费者还是生产者都属于客户端,而我们需要安装RocketMQ,这是属于服务端.和ActivieMQ.zookeeper类似,消费者.生成者.服务端(NameServer)之间是采取观察者模式实现. 在操作系统上安装RocketMQ,启动服务端NameServer.启动Broker,书写Consumer代码,

  • java RocketMQ快速入门基础知识

    如何使用 1.引入 rocketmq-client <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>4.1.0-incubating</version> </dependency> 2.编写Producer DefaultMQProducer produce

  • Docker中RocketMQ的安装与使用详解

    搜索RocketMQ的镜像,可以通过docker的hub.docker.com上进行搜索,也可以在Linux下通过docker的search命令进行搜索,不过最近防火墙升级后,导致国外的网站打开都很慢,通过命令搜索反而会更加方便,操作Docker命令一定要是root用户或者具有root权限的用户.查询操作如下: docker search rocketmq 可以得到如下的结果: 镜像倒是蛮多的,不过看来看去没有一个是官方发布的,我就随便选一个吧,如foxiswho/rocketmq,以下是一个查

  • springBoot整合RocketMQ及坑的示例代码

    版本: JDK:1.8 springBoot:1.5.10 rocketMQ:4.2.0 pom 配置: <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.5.10.RELEASE</version> </parent> <d

  • SpringBoot整合Netty实现WebSocket的示例代码

    目录 一.pom.xml依赖配置 二.代码 2.1.NettyServer 类 2.2.SocketHandler 类 2.3.ChannelHandlerPool 类 2.4.Application启动类 三.测试 一.pom.xml依赖配置 <!-- netty --> <dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <v

  • SpringBoot整合Elasticsearch游标查询的示例代码(scroll)

    游标查询(scroll)简介 scroll 查询 可以用来对 Elasticsearch 有效地执行大批量的文档查询,而又不用付出深度分页那种代价. 游标查询会取某个时间点的快照数据. 查询初始化之后索引上的任何变化会被它忽略. 它通过保存旧的数据文件来实现这个特性,结果就像保留初始化时的索引 视图 一样. 启用游标查询可以通过在查询的时候设置参数 scroll 的值为我们期望的游标查询的过期时间. 游标查询的过期时间会在每次做查询的时候刷新,所以这个时间只需要足够处理当前批的结果就可以了,而不

  • SpringBoot整合Shiro和Redis的示例代码

    demo源码 此demo用SpringBoot+Shiro简单实现了登陆.注册.认证.授权的功能,并用redis做分布式缓存提高性能. 1.准备工作 导入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/XM

  • springboot整合rocketmq实现分布式事务

    1 执行流程 (1) 发送方向 MQ 服务端发送消息. (2) MQ Server 将消息持久化成功之后,向发送方 ACK 确认消息已经发送成功,此时消息为半消息. (3) 发送方开始执行本地事务逻辑. (4) 发送方根据本地事务执行结果向 MQ Server 提交二次确认(Commit 或是 Rollback),MQ Server 收到Commit 状态则将半消息标记为可投递,订阅方最终将收到该消息:MQ Server 收到 Rollback 状态则删除半消息,订阅方将不会接受该消息. (5)

  • springboot 使用自定义的aspect的示例代码

    对某个类型中的方法进行拦截,然后加入固定的业务逻辑,这是AOP面向切面编程可以做的事,在springboot里实现aop的方法也有很多, spring-boot-starter-aop 或者 aspectjweaver 都是可以实现的,不过我们在实现之前,先来看一下aop里的几个概念. 概念 切面(Aspect):是指横切多个对象的关注点的一个模块化,事务管理就是J2EE应用中横切关注点的很好示例.在Spring AOP中,切面通过常规类(基本模式方法)或者通过使用了注解@Aspect的常规类来

  • SpringBoot 签到奖励实现方案的示例代码

    前言 最近在做社交业务,用户进入APP后有签到功能,签到成功后获取相应的奖励: 项目状况:前期尝试业务阶段: 特点: 快速实现(不需要做太重,满足初期推广运营即可) 快速投入市场去运营 用户签到: 用户在每次启动时查询签到记录(规则:连续7日签到从0开始,签到过程中有断签从0开始) 如果今日未签到则提示用户可以进行签到 用户签到获取相应的奖励 提到签到,脑海中首先浮现特点: 需要记录每位用户每天的签到情况 查询时根据规则进行签到记录情况 需求&流程设计&技术实现方案 需求原型图 查询签到记

  • SpringBoot整合Mybatis注解开发的实现代码

    官方文档: https://mybatis.org/mybatis-3/zh/getting-started.html SpringBoot整合Mybatis 引入maven依赖 (IDEA建项目的时候直接选就可以了) <dependency> <groupId>org.mybatis.spring.boot</groupId> <artifactId>mybatis-spring-boot-starter</artifactId> <ve

  • Springboot实现XSS漏洞过滤的示例代码

    背景 前阵子做了几个项目,终于开发完毕,进入了测试阶段,信心满满将项目部署到测试环境,然后做了安全测评之后..... ​(什么!你竟然说我代码不安全???) 然后测出了 Xss漏洞 安全的问题 解决方案 场景:可以在页面输入框输入JS脚本, 攻击者可以利用此漏洞执行恶意的代码 ! 问题演示 ​ ​ 所以我们要对于前端传输的参数做处理,做统一全局过滤处理 既然要过滤处理,我们首先需要实现一个自定义过滤器 总共包含以下四部分 XssUtil XssFilterAutoConfig XssHttpSe

  • SpringBoot 整合 Shiro 密码登录的实现代码

    导入依赖(pom.xml) <!--整合Shiro安全框架--> <dependency> <groupId>org.apache.shiro</groupId> <artifactId>shiro-spring</artifactId> <version>1.4.0</version> </dependency> <!--集成jwt实现token认证--> <dependency

随机推荐

其他