1.分布式消息系统介绍

分布式消息系统是基于区块链实现的消息代理服务，它与区块链节点建立连接，将消息数据存储在区块链节点上。面向用户或物联网设备提供消息发布订阅功能。基于区块链可在各节点共同存储消息的特性，可以实现消息的永久存储，可有效避免因为单个节点故障引起的消息丢失，发布到消息系统上的消息支持永久存储，不可篡改，支持事后跟踪和审计。 确保用户与物联网设备通信时消息可到达。
分布式消息系统是去中心化的，更加关注信任，访问安全，数据不可篡改等特性。非常适合应用在跨组织，跨机构的合作。

2.系统架构

分布式消息系统采用MOM架构,MOM是消息中间件。基本思想是A和B两个应用程序之间不直接收发消息。分布式消息系统是面向消息中间件(MOM)提供的以松散耦合的方式集成应用程序的一种机制。它们提供了基于存储和转发的应用程序之间的异步数据发送，使应用程序彼此不直接通信，而是与作为中介的分布式消息系统通信。分布式消息系统提供了有保证的消息发送，应用程序开发人员无需了解远程过程调用（PRC）和网络/通信协议的细节。

分布式消息系统是区块链节点上的代理服务，是消息上链的入口。相比传统消息系统，分布式消息系统是去中心化的，更加关注信任，访问安全，数据不可篡改等特性。非常适合应用在跨组织，跨机构的合作。
消息系统主要分为两个模块，一是消息处理模块，实现消息的发布和订阅。二是消息系统治理模块，实现消息系统的监管和故障检测。

3.接入支持

分布式消息系统提供多种接入方式：
系统中的消息生产者可以是后台服务、前端网页、移动设备、iot设备。

4.系统概念

不同的场景消息可能有不同的定义，机构之间存在的业务数据和物联网设备通信数据。

机构和机构之间的业务数据信息

场景举例：公路云和虚拟设计院组成一个联盟链。他们属于不同的机构，但他们之间有业务数据往来。在这个场景下这两个机构之间就可以通过联盟链发布和订阅数据。

物联网设备通信

客户端程序通过MQTT(消息队列遥测传输)协议连接分布式消息系统服务。传输和物联网设备的通信数据。

消息主题
主题是区分业务类型的标志，在系统中主题由代理服务和业务名称共同确定。

消息对象
在系统中消息就是消息生产者发生的一条业务数据。消息包含的字段：
消息ID、消息创建时间、消息内容、消息发布者、关联的主题。

发布消息
消息生产者，往某个主题发布消息，消息会被永久保存在区块链上，不可篡改，支持后期审核。

订阅消息
消息消费者首先订阅消息生产者发布的主题。
消费者订阅某个主题后，当生产者往主题发布消息的时候，消费者会及时收到消息。

取消订阅
消费者通过订阅ID取消订阅对应的主题，之后再也收不到对应的主题消息。

5.系统搭建

1.搭建区块链创建共识节点。
2.搭建分布式消息系统 。
3.消息系统和区块链节点通过身份认证机制建立连接。

6.接入说明

提供了RESTful、JsonRPC、STOMP风格的协议接入，为方便物联网IOT设备的使用，提供了内置MQTT协议的支持。 同时为Java业务接入提供了Java SDK客户端。

RESTful

RESTful是一种在Web应用上非常流行的软件架构风格、设计风格，基于HTTP，可以使用XML格式定义或JSON格式定义。基于这个风格设计的软件可以更简洁，REST 接口可以直接在浏览器上测试，给开发和测试过程带来很多便利。
协议说明

• HTTP方法可以使用GET和POST，并且统一使用UTF8字符集。
• 应答的HTTP状态码是 200，输出参数统一是json格式，Content-type: application/json。

接口说明
创建主题

curl "http://localhost:8080/weevent-broker/rest/open?topic=com.weevent.test&groupId=1"

关闭主题

curl "http://localhost:8080/weevent-broker/rest/close?topic=com.weevent.test&groupId=1"

订阅主题

curl http://localhost:8080/weevent/rest/subscribe?topic=com.weevent.test&groupId=1&subscriptionId=c8a600c0-61a7-4077-90f6-3aa39fc9cdd5&url=http%3a%2f%2flocalhost%3a8080%2fweevent%2fmock%2frest%2fonEvent

取消订阅

curl http://localhost:8080/weevent/rest/unSubscribe?subscriptionId=c8a600c0-61a7-4077-90f6-3aa39fc9cdd5

发布消息

curl http://localhost:8080/weevent/rest/publish?topic=com.weevent.test&groupId=1&content=123456&weevent-format=json

接收消息

curl http://localhost:8080/weevent/rest/subscribe?topic=com.weevent.test&groupId=1&subscriptionId=c8a600c0-61a7-4077-90f6-3aa39fc9cdd5&url=http%3a%2f%2flocalhost%3a8080%2fweevent%2fmock%2frest%2fonEvent

获取消息详情

curl http://localhost:8080/weevent/rest/getEvent?eventId=2cf24dba-59-1124&groupId=1

获取主题列表

curl http://localhost:8080/weevent/rest/list?pageIndex=1&pageSize=10&groupId=1

获取主题订阅者列表

curl http://localhost:8080/weevent/admin/listSubscription

stomp协议

STOMP即Simple (or Streaming) Text Orientated Messaging Protocol，简单(流)文本定向消息协议，它提供了一个可互操作的连接格式，允许STOMP客户端与任意STOMP消息代理（Broker）进行交互。STOMP协议由于设计简单，易于开发客户端，因此在多种语言和多种平台上得到广泛地应用。
多语言stomp客户端
接口说明
1.建立连接

// standard web socket transportWebSocketClient webSocketClient = new StandardWebSocketClient();WebSocketStompClient stompClient = new WebSocketStompClient(webSocketClient);// MappingJackson2MessageConverterstompClient.setMessageConverter(new StringMessageConverter());stompClient.setTaskScheduler(taskScheduler); // for heartbeatsListenableFuture<StompSession> f = stompClient.connect("ws://localhost:8080/weevent/stomp", getWebsocketSessionHandlerAdapter());StompSession stompSession = f.get();

2.发布消息

 StompHeaders header = new StompHeaders();header.setDestination("message");header.set("groupId","1");header.set("message-format", "json")StompSession.Receiptable receiptable = stompSession.send(header, "{\"hello\":\" wolrd\"}");log.info("send result, receipt id: {}", receiptable.getReceiptId());

3.订阅消息

StompHeaders header = new StompHeaders();header.setDestination(topic);header.set("messageId","2cf24dba-59-1124");header.set("groupId","1");StompSession.Subscription subscription = stompSession.subscribe(header, new StompFrameHandler() {@Overridepublic Type getPayloadType(StompHeaders headers) {return String.class;}@Overridepublic void handleFrame(StompHeaders headers, Object payload) {logger.info("subscribe handleFrame, header: {} payload: {}", headers, payload);}});

MQTT协议

实现MQTT协议需要客户端和服务器端通讯完成，在通讯过程中，MQTT协议中有三种身份：发布者（Publish）、代理（Broker）服务器、订阅者（Subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。

1.服务配置

#mqtt brokerserver
mqtt.broker.port=8091
mqtt.broker.keepalive=60
mqtt.websocket.path=/weevent/mqtt
mqtt.websocket.port=8092
mqtt.user.login=
mqtt.user.passcode=

2.客户端与分布式消息系统建立连接

const std::string SERVER_ADDRESS("tcp://39.105.47.33:7001/mqtt");
const std::string CLIENT_ID("async_subcribe_cpp");
const std::string TOPIC("message_test");
const std::string mqtt_user = "test";
const std::string mqtt_passwd = "123456";
const int	_QOS = 1;
const int	N_RETRY_ATTEMPTS = 5;mqtt::connect_options connOpts(mqtt_user,mqtt_passwd);connOpts.set_connect_timeout(5);connOpts.set_automatic_reconnect(true);connOpts.set_keep_alive_interval(20);connOpts.set_clean_session(true);mqtt::async_client client(SERVER_ADDRESS, CLIENT_ID);callback cb(client, connOpts);client.set_callback(cb);

3.客户端接收消息

// Callback for when a message arrives.void message_arrived(mqtt::const_message_ptr msg) override {std::cout << "Message arrived" << std::endl;std::cout << "\ttopic: '" << msg->get_topic() << "'" << std::endl;std::cout << "\tpayload: '" << msg->to_string() << "'\n" << std::endl;unsigned char byte_arr[256];int i = 0;Json::Reader reader;Json::Value root;if (reader.parse(msg->to_string(), root))  // reader将Json字符串解析到root，root将包含Json里所有子元素  {std::string even_id = root["eventId"].asString();get_message(even_id.c_str());std::cout << "eventId:" << even_id << std::endl;}

7.应用场景

单机构的场景：

消息在机构内部传输，机构在区块链节点上搭建自己属于自己的分布式消息系统代理服务，消息生产者通过本服务发布消息，消息接受者同时在该服务订阅消息。这样就可以实现消息在机构内部传输。

单机构价值:

可实现消息的在区块链上的永久存储，不会因为系统故障引起数据丢失，降低数据丢失风险。（当属于自己的节点故障，数据也会在共识的区块链节点上存储，当节点恢复数据也同时可以恢复）

多机构之间消息传输：
多个机构在自己区块链节点上搭建消息系统代理服务，各个机构通过消息代理服务将消息发布到区块链上，订阅方就可以接收到对应发布方发送的消息。

价值：
实现数据不可篡改，因为数据存在于链上，对于单个节点，他没有权利更改数据，提供一对多的消息发布，解除应用程序耦合。

消息系统使用案例

各组织可以实现自己的客户端程序，只要支持其中的一种接入协议标准就可以分布式消息系统，实现消息的发布和订阅。这样可以适应在不同端的消息展示风格。

物联网设备使用案例

“物联网+施工”让施工现场也进入“智慧”时代，分布式消息系统可以使远程管理平台获得实时的现场环境数据。
施工现场通过各种传感器采集数据，通过网络传送给分布式消息系统的对应主题，消息订阅方通过订阅传感器的主题来获得传感器数据，再将获得的传感器数据通过客户程序传送到对应的设备上，可以是手机，电脑LED显示屏展示。

施工现场可检测参数有:
①PM2.5、PM10，总悬浮颗粒物TSP(PM100)、噪声
②温度、湿度、气压、风速、风向等气象参数

现场LED显示屏实时显示传感器数值；选配摄像头，对关键作业场所实时视频监控；数据无线远传至远程平台，可选配有线或者无线模式；可对接喷淋设备，自动喷雾降尘。

分布式消息系统因为他支持MQTT协议，所以他可以接入物联网设备。