python关于socket通讯及mqtt通讯

因项目需求，需要在服务机完成计算任务，之后通过IP通讯将数据发送

均为python程序

import socket

发送端代码：

def send_socket():
    server_send = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    while True:
        try:
            server_send.connect((server_ip, server_port))
            print("INFO", "服务器已连接")
            flag = True
        except Exception as e:
            print("ERROR", "连接服务器失败")
            print("ERROR", e)
            flag = False
        while flag:
            try:
                send_data = q_result_socket.get()
                server_send.sendall(send_data.encode())
                print("DEBUG", "数据已发送")
            except Exception as e:
                print("WARNING", "数据发送失败")
                print("WARNING", e)
                break
            # flag = server_send. 
        print("WARNING", "服务器连接丢失，正在重新连接")

接受端代码：

def receive_socket():
    server_receive = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    while True:
        try:
            server_receive.bind((server_ip, server_port))
            server_receive.listen(5)
            conn, addr = server_receive.accept()
            print('服务端已启动')
        except Exception as e:
            print('服务端启动失败', e)
            time.sleep(1)
            continue
        while True:
            try:
                data = conn.recv(1024)
                if not data:
                    break
                else:
                    print(str(data, encoding="utf-8"))
            except Exception as e:
                print(e)
        conn.close()

pip install paho-mqtt

import paho.mqtt.client as mqtt

mqtt通讯测试必须搭建服务器才可以连接，为了测试程序是否可用，故先在自己笔记本安装服务端程序

服务器软件有EMQX，apache-apollo等，不过本人尝试EMQX数个版本均未成功，最后选择apache-apollo。

但apache-apollo需要JAVA才可以，而且高版本JAVA也会出BUG，故安装了javav8。

apache-apollo下载地址：http://archive.apache.org/dist/activemq/activemq-apollo/1.7.1/

JAVA下载地址：https://www.oracle.com/java/technologies/downloads/#java8-windows

安装过程参考链接：https://blog.csdn.net/Omega_Alpha/article/details/99855896（感谢作者）

具体安装过程不再赘述：

下面是通讯的发送端代码：

def send_mqtt():
    print("INFO", "开始启动发送")
    while True:
        flag = False
        client_send = mqtt.Client('computer_0', transport='tcp')
        client_send.username_pw_set(server_user, server_password)
        try:
            client_send.connect(server_ip, server_port, 60)
            client_send.loop_start()
            print("INFO", "服务器已连接")
            flag = True
        except Exception as e:
            print("ERROR", "连接服务器失败")
            print("ERROR", e)
        while flag:
            try:
                send_data = q_result_socket.get()
                info = client_send.publish(server_topic, send_data)
                print("DEBUG", "数据已发送")
            except Exception as e:
                print("WARNING", "数据发送失败")
                print("WARNING", e)
            flag = client_send.is_connected()
        print("WARNING", "服务器连接丢失，正在重新连接")
        client_send.loop_stop()
        time.sleep(1)

订阅端代码：

def on_connect(client, userdata, flags, rc):
    print('connected to mqtt with resurt code ', rc)
    client.subscribe('data')


def on_message(client, userdata, msg):
    payload = msg.payload.decode('utf-8')
    print(payload)


def recv_mqtt():
    client = mqtt.Client(client_id, transport='tcp')
    client.username_pw_set('admin', 'password')
    client.on_connect = on_connect  # 启用订阅模式
    client.on_message = on_message  # 接收消息
    client.connect("192.168.137.1", 61613, 60)  # 链接
    # client.loop_start()   # 以start方式运行，需要启动一个守护线程，让服务端运行，否则会随主线程死亡
    client.loop_forever()    # 以forever方式阻塞运行


if __name__ == '__main__':
    recv_mqtt()