轻量消息队列（原MNS）消费Demo

相关下载

注意事项

在使用示例Demo时，请注意如下信息。

• 配置AccessKey ID和AccessKey Secret信息。

  为避免在代码中硬编码访问密钥（AccessKey）而造成泄露，请通过配置环境变量的方式获取AccessKey。环境变量配置方法，请参见在Linux、macOS和Windows系统配置环境变量。

  本文以环境变量名ALIBABA_CLOUD_ACCESS_KEY_ID和ALIBABA_CLOUD_ACCESS_KEY_SECRET为例，进行后续操作。通过环境变量获取AccessKey的代码示例如下：

  String accessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
  String accessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");

• messageType替换为您需要的消息类型，如订阅WhatsApp通道的上行消息（ChatAppInbound）。Chat App 消息服务支持的回执消息类型，请参见回执消息简介与配置流程。

  String messageType="messageType";

• queueName是消息队列名称，以WhatsApp通道为例，您可以在Chat App 消息服务控制台 > 即时消息 > WhatsApp > 管理 > WABA管理 > 号码管理 > 设置Webhook界面开启消息队列并保存，重新进入设置Webhook界面查看消息队列名称。

  String queueName="queueName";

• Java示例代码中您获取到的上行信息内容由dealMessage方法处理，您可以将需要的上行信息内容业务逻辑写在该方法中。arg代表回执消息体参数，可填写的值如：start_time、end_time、duration、status_code等，建议您根据具体场景填写参数。

  // 根据文档中具体的消息格式进行消息体的解析
  String arg = (String) contentMap.get("arg");
  // 编写您的业务代码

示例Demo

package com.alicom.mns.sample;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import com.alicom.mns.tools.DefaultAlicomMessagePuller;
import com.alicom.mns.tools.MessageListener;
import com.aliyun.mns.model.Message;
import com.google.gson.Gson;

import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * 只能用于接收云通信的消息，不能用于接收其他业务的消息
 * 短信上行消息接收demo
 */
public class ReceiveDemo {

    private static Logger logger = Logger.getLogger(ReceiveDemo.class.getName());

    static class MyMessageListener implements MessageListener {
        private Gson gson = new Gson();

        @Override
        public boolean dealMessage(Message message) {

            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            //消息的几个关键值
            System.out.println("message receiver time from mns:" + format.format(new Date()));
            System.out.println("message handle: " + message.getReceiptHandle());
            System.out.println("message body: " + message.getMessageBodyAsString());
            System.out.println("message id: " + message.getMessageId());
            System.out.println("message dequeue count:" + message.getDequeueCount());
            System.out.println("Thread:" + Thread.currentThread().getName());
            try {
                Map<String, Object> contentMap = gson.fromJson(message.getMessageBodyAsString(), HashMap.class);

                //TODO 根据文档中具体的消息格式进行消息体的解析
                String arg = (String) contentMap.get("arg");

                //TODO 这里开始编写您的业务代码

            } catch (com.google.gson.JsonSyntaxException e) {
                logger.log(Level.SEVERE, "error_json_format:" + message.getMessageBodyAsString(), e);
                //理论上不会出现格式错误的情况，所以遇见格式错误的消息，只能先delete,否则重新推送也会一直报错
                return true;
            } catch (Throwable e) {
                //您自己的代码部分导致的异常，应该return false,这样消息不会被delete掉，而会根据策略进行重推
                return false;
            }

            //消息处理成功，返回true, SDK将调用MNS的delete方法将消息从队列中删除掉
            return true;
        }

    }

    public static void main(String[] args) throws Exception, ParseException {

        DefaultAlicomMessagePuller puller = new DefaultAlicomMessagePuller();

        //设置异步线程池大小及任务队列的大小，还有无数据线程休眠时间
        puller.setConsumeMinThreadSize(6);
        puller.setConsumeMaxThreadSize(16);
        puller.setThreadQueueSize(200);
        puller.setPullMsgThreadSize(1);
        //和服务端联调问题时开启,平时无需开启，消耗性能
        puller.openDebugLog(false);

        //TODO 此处为从操作系统中获取的AK信息
        String accessKeyId=System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
	String accessKeySecret=System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");

        /*
         * TODO 将messageType和queueName替换成您需要的消息类型名称和对应的队列名称
         * Chat App消息服务下所有的回执消息类型请参见如下地址：
         * https://help.aliyun.com/zh/chatapp/developer-reference/configure-message-receipts
         */

        //此处应该替换成相应产品的消息类型
        String messageType = "messageType";
        //在云通信页面开通相应业务消息后，就能在页面上获得对应的queueName,格式类似：Alicom-Queue-******-ChatAppInbound
        String queueName = "queueName";

        puller.startReceiveMsg(accessKeyId, accessKeySecret, messageType, queueName, new MyMessageListener());
    }


}