传感器上传数据到阿⾥云Iot，然后从阿⾥云Iot传输数据到我的
服务器和数据库
整个系统架构如下：
MQTT 和RocketMQ的区别：
1、传感器采⽤GPRS传输数据，采⽤微消息队列MQTT，微消息队列 MQTT 主要承担移动端连接接⼊、连接管理、数据转发等⼯作，相当于⼀个⽆限扩展能⼒的连接⽹关。传感器传送数据到阿⾥云Iot。MQTT⽀持Python，java，有相应的sdk，地址如下：
pink是什么意思
阿⾥云Iot的设备传输数据格式是json格式，采⽤post⽅式提交数据，具体格式如下：
设备属性上报
通过该Topic获取设备上报的属性信息。
Topic：/sys/{productKey}/{deviceName}/thing/event/property/post
数据格式：
{
"iotId":"4z819VQHk6VSLmmBJfrf00107ee200",
"productKey":"1234556554",
"deviceName":"deviceName1234",
"gmtCreate":1510799670074,
"deviceType":"Ammeter",
guilty是什么意思"items":{
"Power":{
"value":"on",
"time":1510799670074
},
"Position":{
"time":1510292697470,
"value":{
july的音标"latitude":39.9,
"longitude":116.38
}
}
}
}
参数说明：
参数类型说明
iotId String设备在平台内的唯⼀标识
productKey String设备所属产品的唯⼀标识
deviceName String设备名称
deviceType String设备类型
items Object设备数据
Power String属性名称，产品所具有的属性名称请参考TSL描述
Position String属性名称，产品所具有的属性名称请参考TSL描述
value根据TSL定义属性值
time Long属性产⽣时间，如果设备没有上报默认采⽤云端⽣成时间gmtCreate Long数据流转消息产⽣时间
发送⽅式如下：
有道翻译官MQTT采⽤topic形式发送数据，⽣产者产⽣数据，通过topic进⾏发送，消费者订阅消息，接收消息。
belly
RocketMQ是⼀个消息队列，吞吐性能强⼤。格式如下，
⾃定义⽣产者与消费者如下：
producer
发送消息
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException { DefaultMQProducer producer = new DefaultMQProducer("pay_topic_01");
producer.tNamesrvAddr("100.8.8.88:9876");
producer.start();
for (int i = 0; i < 1000; i++) {
try {
Message msg = new Message("TopicTest",// topic
"TagA",// tag
("Hello RocketMQ " + i).getBytes()// body
);
SendResult ndResult = producer.nd(msg);
System.out.println(ndResult);
}
catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}
producer.shutdown();
}
}
订阅消息
public class Consumer {
public static void main(String[] args) throws InterruptedException, MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("plea_rename_unique_group_name_4"); consumer.tNamesrvAddr("100.8.8.88:9876");
/**
3s lady* 设置Consumer第⼀次启动是从队列头部开始消费还是队列尾部开始消费<br>
* 如果⾮第⼀次启动，那么按照上次消费的位置继续消费
*/
consumer.tConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("TopicTest", "*");
@Override
dvi d
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,governmental
ConsumeConcurrentlyContext context) {
System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started.");
}
3、阿⾥云官⽅Python 的sdk中的producer和consumer如下：
（1）producer
#初始化 client
mq_client = MQClient(
#设置HTTP接⼊域名（此处以公共云⽣产环境为例）
"${HTTP_ENDPOINT}",
#AccessKey 阿⾥云⾝份验证，在阿⾥云服务器管理控制台创建
"${ACCESS_KEY}",
#SecretKey 阿⾥云⾝份验证，在阿⾥云服务器管理控制台创建
"${SECRET_KEY}"
)
#所属的 Topic
topic_name = "${TOPIC}"
#Topic所属实例ID，默认实例为空None
instance_id = "${INSTANCE_ID}"
just for fun
producer = _producer(instance_id, topic_name)
# 循环发布多条消息
msg_count = 100
print "%sPublish Message To %s\nTopicName:%s\nMessageCount:%s\n" % (10 * "=", 10 * "=", topic_name, msg_count)
try:
for i in range(msg_count):
unconditional
msg_body = "I am test message %s." % i
msg = TopicMessage(
# 消息内容
"I am test message %s." % i,
# 消息标签
""
)
re_msg = producer.publish_message(msg)
print "Publish Message Succeed. MessageID:%s, BodyMD5:%s" % (ssage_id, ssage_body_md5) time.sleep(1)
except MQExceptionBa, e:
pe == "TopicNotExist":
print "Topic not exist, plea create it."
print "Publish Message Fail. Exception:%s" % e
（2）consumer
重要的是message的body中的数据，取出数据即可实现通讯。
#初始化 client
mq_client = MQClient(
#设置HTTP接⼊域名（此处以公共云⽣产环境为例）
"${HTTP_ENDPOINT}",
#AccessKey 阿⾥云⾝份验证，在阿⾥云服务器管理控制台创建
"${ACCESS_KEY}",
#SecretKey 阿⾥云⾝份验证，在阿⾥云服务器管理控制台创建
"${SECRET_KEY}"
)
#所属的 Topic
topic_name = "${TOPIC}"
#您在控制台创建的 Consumer ID(Group ID)
group_id = "${GROUP_ID}"
#Topic所属实例ID，默认实例为空None
instance_id = "${INSTANCE_ID}"
consumer = _consumer(instance_id, topic_name, group_id)
