Springboot如何实现RocketMq

小编这次要给大家分享的是Springboot如何实现RocketMq，文章内容丰富，感兴趣的小伙伴可以来了解一下，希望大家阅读完这篇文章之后能够有所收获。

创新互联公司专注为客户提供全方位的互联网综合服务，包含不限于网站建设、做网站、三台网络推广、小程序设计、三台网络营销、三台企业策划、三台品牌公关、搜索引擎seo、人物专访、企业宣传片、企业代运营等，从售前售中售后，我们都将竭诚为您服务，您的肯定，是我们最大的嘉奖；创新互联公司为所有大学生创业者提供三台建站搭建服务，24小时服务热线：18980820575，官方网址：www.cdcxhl.com

首先，在虚拟机上安装rocketmq和rocketMq可视化控制，安装不做描述。

1、pom.xml文件添加依赖

mq的版本与连接的rocketmq版本保持一致

    
      org.apache.rocketmq
      rocketmq-remoting
      4.4.0

2、yml文件添加rocketmq配置

apache:
 rocketmq:
  #消费者的配置
  consumer:
   pushConsumer: myConsumer
  #生产者的配置
  producer:
   producerGroup: myGroup
  namesrvAddr: 192.168.233.128:9876

3、生产者类RocketProducer

package com.zp.springbootdemo.rocketmq;

import com.alibaba.fastjson.JSONObject;
import com.sun.org.apache.xpath.internal.objects.XString;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
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.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;

/**
 * @Author zp
 * @Description rocketmq生产者
 * @Date 22:06 2020/5/22
 * @Param
 * @return
 **/
@Component
public class RocketProducer {
  /**
   * 生产者的组名
   */
  @Value("${apache.rocketmq.producer.producerGroup}")
  private String producerGroup;
  /**
   * NameServer 地址
   */
  @Value("${apache.rocketmq.namesrvAddr}")
  private String namesrvAddr;

  private DefaultMQProducer defaultMQProducer;

  @PostConstruct
  public void defaultMQProducer(){
    //生产者的组名
    defaultMQProducer = new DefaultMQProducer(producerGroup);
    defaultMQProducer.setNamesrvAddr(namesrvAddr);
    defaultMQProducer.setVipChannelEnabled(false);
    try {
      defaultMQProducer.start();
      System.out.println("producer启动了。。。");
    } catch (MQClientException e) {
      e.printStackTrace();
    }
  }

  public String send(String topic,String tags,String body) throws UnsupportedEncodingException, InterruptedException, RemotingException, MQClientException, MQBrokerException {
    Message message = new Message(topic,tags,body.getBytes(RemotingHelper.DEFAULT_CHARSET));
    StopWatch stop = new StopWatch();
    stop.start();
    SendResult result = defaultMQProducer.send(message);
    System.out.println("发送响应：MsgId:" + result.getMsgId() + "，发送状态:" + result.getSendStatus());
    JSONObject jsonObject = new JSONObject();
    jsonObject.put("msgId",result.getMsgId());
    jsonObject.put("sendStatus",result.getSendStatus());
    stop.stop();
    return jsonObject.toJSONString();
  }
}

4、消费者类RocketConsumer

package com.zp.springbootdemo.rocketmq;

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.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

/**
 * @Author zp
 * @Description rocketmq消费者
 * @Date 22:33 2020/5/22
 * @Param
 * @return
 **/
@Component
public class RockerConsumer implements CommandLineRunner {
  /**
   * 消费者
   */
  @Value("${apache.rocketmq.consumer.pushConsumer}")
  private String pushConsumer; //myConsumer

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

  /**
   * 初始化RocketMq的监听信息，渠道信息
   */
  public void messageListener(){
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(pushConsumer);
    consumer.setNamesrvAddr(namesrvAddr);

    try {
      // 订阅PushTopic下Tag为push的消息,都订阅消息
      consumer.subscribe("firstTopic","push");
      // 程序第一次启动从消息队列头获取数据
      consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
      //可以修改每次消费消息的数量，默认设置是每次消费一条
      consumer.setConsumeMessageBatchMaxSize(1);

      //在此监听中消费信息，并返回消费的状态信息
      consumer.registerMessageListener((MessageListenerConcurrently) (msgs,context)->{
        // 会把不同的消息分别放置到不同的队列中
        for (Message msg:msgs){
          System.out.println("接收到了消息："+new String(msg.getBody()));

        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      });
      consumer.start();
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
  /**
   * Callback used to run the bean.
   *
   * @param args incoming main method arguments
   * @throws Exception on error
   */
  @Override
  public void run(String... args) throws Exception {
    this.messageListener();
  }
}

5、controller中编写发送消息

package com.zp.springbootdemo.rocketmq;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.UnsupportedEncodingException;

@RestController
@RequestMapping("/rocketMq")
public class MQController {

  @Autowired
  private RocketProducer producer;

  @RequestMapping("/myFirstProducer")
  public String pushMsg(String msg){
    try {
      System.out.println("======"+msg);
      return producer.send("firstTopic","push",msg);
    } catch (InterruptedException e) {
      e.printStackTrace();
    } catch (RemotingException e) {
      e.printStackTrace();
    } catch (MQClientException e) {
      e.printStackTrace();
    } catch (MQBrokerException e) {
      e.printStackTrace();
    } catch (UnsupportedEncodingException e) {
      e.printStackTrace();
    }
    return "ERROR";
  }
}

6.测试

请求地址：http://127.0.0.1:8080/rocketMq/myFirstProducer?msg=hello

响应：{"msgId":"C0A8010E1A3818B4AAC2711E8CD50000","sendStatus":"SEND_OK"}

通过rocketMq可视化控制查看：

Springboot如何实现RocketMq

看完这篇关于Springboot如何实现RocketMq的文章，如果觉得文章内容写得不错的话，可以把它分享出去给更多人看到。

名称栏目：Springboot如何实现RocketMq
本文URL：http://pwwzsj.com/article/pseipj.html

平武建站

Springboot如何实现RocketMq

其他资讯