ONS发布订阅消息

ONS， 全名Open Notification Service, 是阿里基于开源消息中间件RocketMQ的一个云产品。

首先，要申请阿里账号等。本地也可以申请阿里云账号自己调试。此处为公司拥有阿里云服务，与运维op申请ons测试可用的服务。

申请ONS需要的创建topic，PID，CID，access_key，secret_key，要给产生access_key的账户授权发布订阅功能权限，具体请看其他配置阿里云账户权限的文章。

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

使用ONS API发送订阅消息的示例

首先，pom引用ons-client

        <!-- RocketMq ons-client-->

        <dependency>

            <groupId>com.aliyun.openservices</groupId>

            <artifactId>ons-client</artifactId>

            <version>1.7.0.Final</version>

        </dependency>

第二步，配置消费者：

• 创建一个常量类存放

/**

 * 公共参数配置

 *

 */

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";

}

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

import com.aliyun.openservices.ons.api.*;

import java.util.Date;

import java.util.Properties;

public class ONSUtil {

    /**

     * 获取消息的 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;

    }

}　

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

import com.aliyun.openservices.ons.api.Message;

import com.aliyun.openservices.ons.api.Producer;

import com.aliyun.openservices.ons.api.SendResult;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.stereotype.Service;

@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 = ONSUtil.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 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 = ONSUtil.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());

    }

}

    @RequestMapping(value = "/testmqProducer", method = RequestMethod.GET)

    public void testmq() {

         OnsProducer onsProducer=new OnsProducer();

         String msg = "testmq";

         onsProducer.sendMessage(MqConfigParams.TOPIC,"*",msg);

    }

接下来写消费者示例：

• 创建监听类实现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;

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 java.util.Date;

import java.util.Properties;

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启动类中加入消费者启动调用方法

@SpringBootApplication

@MapperScan("com.ons.test")

public class Application {

    public static void main(String[] args) {

        // 启动嵌入式的 Tomcat 并初始化 Spring 环境及其各 Spring 组件

        SpringApplication.run(Application.class,args);

        new MyMessageConsumer().subscribe(); // 系统启动的时候启动订阅

    }}