关于MNS消息发送和接收的实现问题

我在使用MNS发送消息时，用的是阿里云的MNS。一些问题大都可以在官方文档上找到。

消息服务MNS提供了两种API接口，一种是队列接口，一种是主题接口。

队列接口适用于点对点的消息收发，当接收消息时，需要应用端自行轮询获取消息（拉模式）。

主题接口适用于一对多的消息收发，应用端只需要在某个地址上启动监听，服务端就会主动将消息推送过去（推模式）。

1，其实使用两种方式都有自己的利弊，不然提供两种方式也就没有什么区别了。

2，队列模型保证消息至少会被消费一次，支持多个生产者和消费者并发操作同一个消息队列。消费消息时尽量做到先进先出，正是因为分布式消息队列的一些特性并不能保证你能按照消息的发送顺序消费消息，如果你的业务必需先进先出，建议在消息中加入序号信息以便消费消息后进行重新排序。

3，主题模型支持服务端主动将消息推送给用户指定的回调地址（Endpoint），消除用户程序不必要的轮询和资源消耗。
主题模型保证通知消息按照指定的策略推送给用户，支持多个消息发布者并发操作同一个主题。主题模式支持一对多广播消息，一条通知消息可以同时被多个订阅者接收和消费。

我这里主要说下实现方法：

一、队列接口

1.创建队列

public class CreateQueueDemo {public static void main(String[] args) {CloudAccount account = new CloudAccount("YourAccessId", "YourAccessKey", "MNSEndpoint");MNSClient client = account.getMNSClient(); // 在程序中，CloudAccount以及MNSClient单例实现即可，多线程安全String queueName = "TestQueue";QueueMeta meta = new QueueMeta(); //生成本地QueueMeta属性，有关队列属性详细介绍见https://help.aliyun.com/document_detail/27476.htmlmeta.setQueueName(queueName);  // 设置队列名meta.setPollingWaitSeconds(15);meta.setMaxMessageSize(2048L);try {CloudQueue queue = client.createQueue(meta);} catch (ClientException ce){System.out.println("Something wrong with the network connection between client and MNS service."+ "Please check your network and DNS availablity.");ce.printStackTrace();} catch (ServiceException se){se.printStackTrace();logger.error("MNS exception requestId:" + se.getRequestId(), se);if (se.getErrorCode() != null) {if (se.getErrorCode().equals("QueueNotExist")){System.out.println("Queue is not exist.Please create before use");} else if (se.getErrorCode().equals("TimeExpired")){System.out.println("The request is time expired. Please check your local machine timeclock");}/*you can get more MNS service error code from following link:https://help.aliyun.com/document_detail/mns/api_reference/error_code/error_code.html*/}} catch (Exception e){System.out.println("Unknown exception happened!");e.printStackTrace();}client.close();  // 程序退出时，需主动调用client的close方法进行资源释放}
}

当然，你必须有阿里云提供的相关的key、参数和密钥。

2.发送消息

public class ProducerDemo {public static void main(String[] args) {CloudAccount account = new CloudAccount("YourAccessId", "YourAccessKey", "MNSEndpoint");MNSClient client = account.getMNSClient(); // 在程序中，CloudAccount以及MNSClient单例实现即可，多线程安全try {CloudQueue queue = client.getQueueRef("TestQueue");Message message = new Message();message.setMessageBody("I am test message ");Message putMsg = queue.putMessage(message);System.out.println("Send message id is: " + putMsg.getMessageId());} catch (ClientException ce){System.out.println("Something wrong with the network connection between client and MNS service."+ "Please check your network and DNS availablity.");ce.printStackTrace();} catch (ServiceException se){se.printStackTrace();logger.error("MNS exception requestId:" + se.getRequestId(), se);if (se.getErrorCode() != null) {if (se.getErrorCode().equals("QueueNotExist")){System.out.println("Queue is not exist.Please create before use");} else if (se.getErrorCode().equals("TimeExpired")){System.out.println("The request is time expired. Please check your local machine timeclock");}/*you can get more MNS service error code from following link:https://help.aliyun.com/document_detail/mns/api_reference/error_code/error_code.html*/}} catch (Exception e){System.out.println("Unknown exception happened!");e.printStackTrace();}client.close();  // 程序退出时，需主动调用client的close方法进行资源释放}
}

3.接受和删除消息

public class ConsumerDemo {public static void main(String[] args) {CloudAccount account = new CloudAccount("YourAccessId", "YourAccessKey", "MNSEndpoint");MNSClient client = account.getMNSClient(); // 在程序中，CloudAccount以及MNSClient单例实现即可，多线程安全try{CloudQueue queue = client.getQueueRef("TestQueue");Message popMsg = queue.popMessage();if (popMsg != null){System.out.println("message handle: " + popMsg.getReceiptHandle());System.out.println("message body: " + popMsg.getMessageBodyAsString());System.out.println("message id: " + popMsg.getMessageId());System.out.println("message dequeue count:" + popMsg.getDequeueCount());//删除已经取出消费的消息queue.deleteMessage(popMsg.getReceiptHandle());System.out.println("delete message successfully.\n");}else{System.out.println("message not exist in TestQueue.\n");}} catch (ClientException ce){System.out.println("Something wrong with the network connection between client and MNS service."+ "Please check your network and DNS availablity.");ce.printStackTrace();} catch (ServiceException se){se.printStackTrace();logger.error("MNS exception requestId:" + se.getRequestId(), se);if (se.getErrorCode() != null) {if (se.getErrorCode().equals("QueueNotExist")){System.out.println("Queue is not exist.Please create before use");} else if (se.getErrorCode().equals("TimeExpired")){System.out.println("The request is time expired. Please check your local machine timeclock");}/*you can get more MNS service error code from following link:https://help.aliyun.com/document_detail/mns/api_reference/error_code/error_code.html*/}} catch (Exception e){System.out.println("Unknown exception happened!");e.printStackTrace();}client.close();}
}

4.删除队列

public class DeleteQueueDemo {public static void main(String[] args) {CloudAccount account = new CloudAccount("YourAccessId", "YourAccessKey", "MNSEndpoint");MNSClient client = account.getMNSClient(); // 在程序中，CloudAccount以及MNSClient单例实现即可，多线程安全try{CloudQueue queue = client.getQueueRef("TestQueue");queue.delete();} catch (ClientException ce){System.out.println("Something wrong with the network connection between client and MNS service."+ "Please check your network and DNS availablity.");ce.printStackTrace();} catch (ServiceException se){se.printStackTrace();} catch (Exception e){System.out.println("Unknown exception happened!");e.printStackTrace();}client.close();}
}

二、主题方式

1.创建主题

public class CreateTopicDemo {public static void main(String[] args) {CloudAccount account = new CloudAccount("YourAccessId", "YourAccessKey", "MNSEndpoint");MNSClient client = account.getMNSClient(); // 在程序中，CloudAccount以及MNSClient单例实现即可，多线程安全String topicName = "TestTopic";TopicMeta meta = new TopicMeta();meta.setTopicName(topicName);try {CloudTopic topic = client.createTopic(meta);} catch (Exception e)e.printStackTrace();System.out.println("create topic error, " + e.getMessage());}client.close();}
}

2.发送消息

public class PublishMessageDemo {public static void main(String[] args) {CloudAccount account = new CloudAccount("YourAccessId", "YourAccessKey", "MNSEndpoint");MNSClient client = account.getMNSClient(); // 在程序中，CloudAccount以及MNSClient单例实现即可，多线程安全CloudTopic topic = client.getTopicRef("TestTopic");try {TopicMessage msg = new Base64TopicMessage(); //可以使用TopicMessage结构，选择不进行Base64加密msg.setMessageBody("hello world!");msg.setMessageTag("filterTag"); //设置该条发布消息的filterTagmsg = topic.publishMessage(msg);System.out.println(msg.getMessageId());System.out.println(msg.getMessageBodyMD5());} catch (Exception e) {e.printStackTrace();System.out.println("subscribe error");}client.close();}
}

3.创建订阅、接受消息

public class ConsumerDemo {public static void main(String[] args) {CloudAccount account = new CloudAccount("YourAccessId", "YourAccessKey", "MNSEndpoint");MNSClient client = account.getMNSClient(); // 在程序中，CloudAccount以及MNSClient单例实现即可，多线程安全try{CloudTopic topic = client.getTopicRef("TestTopic");ubscriptionMeta subMeta = new SubscriptionMeta();subMeta.setSubscriptionName("TestSub");subMeta.setNotifyContentFormat(SubscriptionMeta.NotifyContentFormat.SIMPLIFIED);subMeta.setEndpoint(topic.generateQueueEndpoint("TestQueue");subMeta.setFilterTag("filterTag");topic.subscribe(subMeta);//订阅相关主题CloudQueue queue = client.getQueueRef("TestQueue");Message popMsg = queue.popMessage();if (popMsg != null){System.out.println("message handle: " + popMsg.getReceiptHandle());System.out.println("message body: " + popMsg.getMessageBodyAsString());System.out.println("message id: " + popMsg.getMessageId());System.out.println("message dequeue count:" + popMsg.getDequeueCount());//删除已经取出消费的消息queue.deleteMessage(popMsg.getReceiptHandle());System.out.println("delete message successfully.\n");}else{System.out.println("message not exist in TestQueue.\n");}} catch (ClientException ce){System.out.println("Something wrong with the network connection between client and MNS service."+ "Please check your network and DNS availablity.");ce.printStackTrace();} catch (ServiceException se){se.printStackTrace();logger.error("MNS exception requestId:" + se.getRequestId(), se);if (se.getErrorCode() != null) {if (se.getErrorCode().equals("QueueNotExist")){System.out.println("Queue is not exist.Please create before use");} else if (se.getErrorCode().equals("TimeExpired")){System.out.println("The request is time expired. Please check your local machine timeclock");}/*you can get more MNS service error code from following link:https://help.aliyun.com/document_detail/mns/api_reference/error_code/error_code.html*/}} catch (Exception e){System.out.println("Unknown exception happened!");e.printStackTrace();}client.close();}
}

4.取消订阅、删除zhu't

队列模型保证消息至少会被消费一次，支持多个生产者和消费者并发操作同一个消息队列。

消费消息时尽量做到先进先出，正是因为分布式消息队列的一些特性并不能保证你能按照消息的发送

顺序消费消息，如果你的业务必需先进先出，建议在消息中加入序号信息以便消费消息后进行重新排

序。

队列模型保证消息至少会被消费一次，支持多个生产者和消费者并发操作同一个消息队列。

消费消息时尽量做到先进先出，正是因为分布式消息队列的一些特性并不能保证你能按照消息的发送

顺序消费消息，如果你的业务必需先进先出，建议在消息中加入序号信息以便消费消息后进行重新排

序。

队列模型保证消息至少会被消费一次，支持多个生产者和消费者并发操作同一个消息队列。

消费消息时尽量做到先进先出，正是因为分布式消息队列的一些特性并不能保证你能按照消息的发送

顺序消费消息，如果你的业务必需先进先出，建议在消息中加入序号信息以便消费消息后进行重新排

序。

队列模型保证消息至少会被消费一次，支持多个生产者和消费者并发操作同一个消息队列。

消费消息时尽量做到先进先出，正是因为分布式消息队列的一些特性并不能保证你能按照消息的发送

顺序消费消息，如果你的业务必需先进先出，建议在消息中加入序号信息以便消费消息后进行重新排

序。

public class DeleteTopicDemo {public static void main(String[] args) {CloudAccount account = new CloudAccount("YourAccessId", "YourAccessKey", "MNSEndpoint");MNSClient client = account.getMNSClient(); // 在程序中，CloudAccount以及MNSClient单例实现即可，多线程安全CloudTopic topic = client.getTopicRef("TestTopic");try {topic.unsubscribe("TestSub");topic.delete();} catch (Exception e) {e.printStackTrace();System.out.println("delete topic error");}client.close();}
}

这个几乎是官网的案例。

但是只要按照这样的顺序和思路写一定不会有问题的。

在使用Spring的时候多使用注入方式。

所以考虑到这个，一半会将这些东西的配置信息写在XML里边。如

    <bean id="queueMeta" class="com.aliyun.mns.model.QueueMeta"><property name="queueName" value="${queueName}"/><property name="pollingWaitSeconds" value="15"/><property name="maxMessageSize" value="2048"/></bean>

关于MNS消息发送和接收的实现问题相关推荐

python 网络编程之Socket通信案例消息发送与接收
背景网络编程是python编程中的一项基本技术.本文将实现一个简单的Socket通信案例消息发送与接收正文在python中的socket编程的大致流程图如上所示我们来首先编写客户端的代码: # ...
go 实现 kafka 消息发送、接收
引言网络上关于 go 实现 kafka 消息发送和接收的文章很多,但是实际操作起来又不是很清楚,本文在网络资源的基础上,结合自己搭建过程中遇到的问题进行了总结. 本文的实验主机:Mac笔记本. 一. ...
使用Akka持久化——消息发送与接收
版权声明:本文为博主原创文章,未经博主允许不得转载. https://blog.csdn.net/beliefer/article/details/53929751 前言在<使用Akka持久化 ...
RabbitMQ消息发送和接收
1.RabbitMQ的消息发送和接受机制所有 MQ 产品从模型抽象上来说都是一样的过程: 消费者(consumer)订阅某个队列.生产者(producer)创建消息,然后发布到队列(queue)中, ...
【转】DICOM医学图像处理：DIMSE消息发送与接收“大同小异”之DCMTK fo-dicom mDCM
转自:https://my.oschina.net/zssure/blog/354816 背景: 从DICOM网络传输一文开始,相继介绍了C-ECHO.C-FIND.C-STORE.C-MOVE等DI ...
JAVA ActiveMQ消息发送和接收
JMS即Java消息服务(Java Message Service)应用程序接口是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信. ...
linux ibm mq 安装，消息发送与接收
下载地址 http://public.dhe.ibm.com/ibmdl/export/pub/software/websphere/messaging/mqadv/ 安装 1.2 解压并安装 1.2 ...
Golang实现Kafka消息发送、接收
一:核心概念 kafka是消息中间件的一种,是一种分布式流平台,是用于构建实时数据管道和流应用程序.具有横向扩展,容错,wicked fast(变态快)等优点. kafka中涉及的名词: 消息记录(r ...
Kafka入门教程 Golang实现Kafka消息发送、接收
一:核心概念 kafka是消息中间件的一种,是一种分布式流平台,是用于构建实时数据管道和流应用程序.具有横向扩展,容错,wicked fast(变态快)等优点. kafka中涉及的名词: 消息记录(r ...

关于MNS消息发送和接收的实现问题

关于MNS消息发送和接收的实现问题相关推荐

最新文章

热门文章