springboot集成阿里ons消息队列发布订阅消息功能

川长思鸟来 2022-05-21 04:12 467阅读 0赞

此处的项目是springboot项目。使用队列的产品是阿里云ons 消息队列

阿里云的ons消息队列是基于rockermq

项目环境。jdk1.8

使用阿里ons开发的api接口实现发布定于功能生产和消费信息

首先，要申请阿里账号等。本地也可以申请阿里云账号自己调试

此处为公司拥有阿里云服务，与运维op申请ons测试可用的服务。

申请ons需要的创建topic，PID,CID,access\_key,secret\_key，要给产生access\_key的账户授权发布订阅功能权限，具体请看其他配置阿里云账户权限的文章

密匙：

![70][]

配置授权策略

![70 1][]

![70 2][]

提供了这些以后，可以在官网查看ons快速入门文档创建简单的demo测试类调试生产和消费

可以访问连接：

[https://help.aliyun.com/document\_detail/34411.html?spm=5176.11065259.1996646101.searchclickresult.1484c2f02oFFz7][https_help.aliyun.com_document_detail_34411.html_spm_5176.11065259.1996646101.searchclickresult.1484c2f02oFFz7]

此处是在你看完文档对ons使用有一个初步了解以后。在springboot项目中配置使用ons api发送订阅消息的示例

首先，pom引入

<dependency>
                <groupId>com.aliyun.openservices</groupId>
                <artifactId>ons-client</artifactId>
                <version>1.7.0.Final</version>
            </dependency>

第二步配置消费者：

2.1创建一个常量类存放

/**
     * 公共参数配置
     * 
     * @author Administrator
     *
     */
    public interface MqConfigParams {
    
        //测试ons配置文件
        public static final String TOPIC = "你申请下来的topic";
        public static final String TAG = "*";
        public static final String PRODUCER_ID = "你申请下来的pid";
        public static final String CONSUMER_ID = "你申请下来的cid";
        public static final String ACCESS_KEY = "你申请下来的accesskey";
        public static final String SECRET_KEY = "你申请下来的secretkey";
        public static final String ONS_ADDR = "http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet";
    
    
    }

2.2创建一个生产者类，此处填入配置文件中的常量类，然后调用getproduce方法时启动生产者 producer.start();

import com.aliyun.openservices.ons.api.*;
    import com.renrenche.databus.common.MqConfigParams;
    
    import java.util.Date;
    import java.util.Properties;
    /**
     * Created by qixin on 2018/6/28.
     */
    public class CloudMQUtil {
    
        /**
         * 获取消息的 Producer
         *
         * @return Producer
         */
        public static Producer getProducer() {
            Properties properties = new Properties();
            properties.put(PropertyKeyConst.ProducerId, MqConfigParams.PRODUCER_ID);
            properties.put(PropertyKeyConst.AccessKey, MqConfigParams.ACCESS_KEY);
            properties.put(PropertyKeyConst.SecretKey, MqConfigParams.SECRET_KEY);
            Producer producer = ONSFactory.createProducer(properties);
    
            // 在发送消息前，必须调用start方法来启动Producer，只需调用一次即可。
            producer.start();
            return producer;
        }
    
    
    
    
    }

2.3 封装一个发送消息的方法。

import com.aliyun.openservices.ons.api.Message;
    import com.aliyun.openservices.ons.api.Producer;
    import com.aliyun.openservices.ons.api.SendResult;
    
    import com.renrenche.databus.seo.controller.SeoLogController;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Service;
    
    /**
     * @author Mood
     * @date: 2018/6/28 15:30
     * @version: 1.0
     * @description: 推送消息给队列通用方法
     */
    @Service
    public class OnsProducer {
    
    
        private final static Logger logger = LoggerFactory.getLogger(SeoLogController.class);
    
        public void sendMessage(String topic, String tag, String msgStr) {
            StringBuilder logsb = new StringBuilder("send OnsMQ Msg:");
            Message msg = getInstance(topic, tag, msgStr);
            String messageId = "";
            Producer producer = CloudMQUtil.getProducer(); //你申请的producerId
            SendResult sendResult = producer.send(msg);
            messageId = sendResult.getMessageId();
            if (messageId != null && !messageId.equals("")) {
                logsb.append("[OnsProducer] : " + "{ messageId : " + messageId + " , msgStr :" + msgStr + "}");
            } else {
                logsb.append("[OnsProducer] fail: " + "{ " + " msgStr :" + msgStr + "}");
            }
            logger.debug(logsb.toString());
        }
    
        private static Message getInstance(String topic, String tag, String body) {
            if (body.equals("")||body == null)
                body = "";
            Message msg = new Message(topic, tag, body.getBytes());
            return msg;
        }
    
    
    }

这样就配置好了生产者。

调用生产者

可以写一个test类用main方法调试

示例这里是直接写main方法执行上面封装的内容直接执行看查看日志看是否成功

import com.aliyun.openservices.ons.api.Message;
    import com.aliyun.openservices.ons.api.Producer;
    import com.aliyun.openservices.ons.api.SendResult;
    import com.renrenche.databus.producer.CloudMQUtil;
    import com.renrenche.databus.common.MqConfigParams;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    public class TestSendMessage {
        private final static Logger logger = LoggerFactory.getLogger(TestSendMessage.class);
        public static void main(String[] args) {
    
            StringBuilder logsb=new StringBuilder("send OnsMQ Msg:");
            Producer producer = CloudMQUtil.getProducer(); //你申请的producerId
            Message msg = new Message(MqConfigParams.TOPIC, //你申请的TopicName
                    "*","这是消息2".getBytes());
            SendResult sendResult = producer.send(msg);
            String messageId = "";
            messageId = sendResult.getMessageId();
            if (messageId != null || !messageId.equals("")) {
                logsb.append("[OnsProducer] : " + "{ messageId : " + messageId + " , msgStr :" + "这是消息2" + "}");
            } else {
                logsb.append("[OnsProducer] fail: " + "{ " + " msgStr :" + "这是消息2" + "}");
            }
            logger.debug(logsb.toString());
        }
    }

我这里是写了一哥testcontroller，用postman调用

@RequestMapping(value = "/testmqProducer", method = RequestMethod.GET)
        public void testmq() {
             OnsProducer onsProducer=new OnsProducer();
             String msg="testmq";
             onsProducer.sendMessage(MqConfigParams.TOPIC,"*",msg);
    
        }

查看打印日志，打印了发送上去的messageId。就证明数据已经生产并发送成功了

![70 3][]

接下来写消费者示例：

第一步创建监听类实现MessageListener此处是处理业务逻辑因为此处可以得到消费的内容，

import com.aliyun.openservices.ons.api.Action;
    import com.aliyun.openservices.ons.api.ConsumeContext;
    import com.aliyun.openservices.ons.api.Message;
    import com.aliyun.openservices.ons.api.MessageListener;
    
    import java.util.Date;
    
    /**
     * Created by qixin on 2018/6/28.
     */
    public class MyMessageListener implements MessageListener {
        @Override
        public Action consume(Message message, ConsumeContext consumeContext) {
            System.out.println("Receive @" + new Date() + ": " + message);
            //此处可以写具体业务逻辑，body是具体发送的内容
            String body = new String(message.getBody());
            System.out.println("msgBody is : " + body);
            return Action.CommitMessage;
        }
    }

第二步：创建消费者启动类和方法

import com.aliyun.openservices.ons.api.Consumer;
    import com.aliyun.openservices.ons.api.ONSFactory;
    import com.aliyun.openservices.ons.api.PropertyKeyConst;
    import com.renrenche.databus.common.MqConfigParams;
    import com.renrenche.databus.listenser.MyMessageListener;
    
    import java.util.Date;
    import java.util.Properties;
    
    /**
     * Created by qixin on 2018/6/28.
     */
    public class MyMessageConsumer {
    
        /**
         * 订阅消息
         */
        public void subscribe() {
            Properties properties = new Properties();
            properties.put(PropertyKeyConst.ConsumerId, MqConfigParams.CONSUMER_ID);
            properties.put(PropertyKeyConst.AccessKey, MqConfigParams.ACCESS_KEY);
            properties.put(PropertyKeyConst.SecretKey, MqConfigParams.SECRET_KEY);
            Consumer consumer = ONSFactory.createConsumer(properties);
            consumer.subscribe(MqConfigParams.TOPIC, "*", new MyMessageListener());//此处可以写父类messageListener但是必须实现方法这里就是用到了上面写的myMessageListener
            consumer.start();
            System.out.println(MqConfigParams.CONSUMER_ID + " is running @" + new Date());
        }
    }

第三步：在Application启动类中加入消费者启动调用方法   
new MyMessageConsumer().subscribe(); // 系统启动的时候启动订阅

@SpringBootApplication
    // mapper 接口类扫描包配置
    @MapperScan("com.renrenche.databus.seo.dao")
    public class Application {
    
        public static void main(String[] args) {
            // 程序启动入口
            // 启动嵌入式的 Tomcat 并初始化 Spring 环境及其各 Spring 组件
            SpringApplication.run(Application.class,args);
            new MyMessageConsumer().subscribe(); // 系统启动的时候启动订阅
        }}

调试：

启动项目，项目启动后消费者就启动了。调用生产者方法发送消息。消费者即刻得到消费消息。

日志查看运行效果

![70 4][]

结束！

感谢观看

[70]: /images/20220521/3107fcbdce374698814bf8e680f68dd6.png
[70 1]: /images/20220521/b650af7a060e4f67a0280064a9873fa8.png
[70 2]: /images/20220521/4fabffe7b1184c5eb7335870f6f0ecc4.png
[https_help.aliyun.com_document_detail_34411.html_spm_5176.11065259.1996646101.searchclickresult.1484c2f02oFFz7]: https://help.aliyun.com/document_detail/34411.html?spm=5176.11065259.1996646101.searchclickresult.1484c2f02oFFz7
[70 3]: /images/20220521/fe80eb7ee229479da37ab3dab3682c72.png
[70 4]: /images/20220521/7ef5e742f68a4202a844be93622e428d.png