这是 Channel 类的一个非常简单的实现。它创建一个套接字,以接受
来自客户端的连接并发送消息。它本身也是一个客户,
从其他 Channel 实例接收消息(例如在单独的进程中)。
通信是在两个线程中完成的,这非常糟糕(我会使用异步io)。什么时候
收到消息,它调用注册的函数在接收线程中哪个
可能会导致一些线程问题。
每个 Channel 实例都会创建自己的套接字,但它的可扩展性要高得多
具有由单个实例复用的通道“主题”。
一些现有的库提供“通道”功能,例如nanomsg http://nanomsg.org.
这里的代码用于教育目的,如果它可以帮助......
import socket
import threading
class ChannelThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.clients = []
self.chan_sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
self.chan_sock.bind(('',0))
_, self.port = self.chan_sock.getsockname()
self.chan_sock.listen(5)
self.daemon=True
self.start()
def run(self):
while True:
new_client = self.chan_sock.accept()
if not new_client:
break
self.clients.append(new_client)
def sendall(self, msg):
for client in self.clients:
client[0].sendall(msg)
class Channel(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.daemon = True
self.channel_thread = ChannelThread()
def public_address(self):
return "tcp://%s:%d" % (socket.gethostname(), self.channel_thread.port)
def register(self, channel_address, update_callback):
host, s_port = channel_address.split("//")[-1].split(":")
port = int(s_port)
self.peer_chan_sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
self.peer_chan_sock.connect((host, port))
self._callback = update_callback
self.start()
def deal_with_message(self, msg):
self._callback(msg)
def run(self):
data = ""
while True:
new_data = self.peer_chan_sock.recv(1024)
if not new_data:
# connection reset by peer
break
data += new_data
msgs = data.split("\n\n")
if msgs[-1]:
data = msgs.pop()
for msg in msgs:
self.deal_with_message(msg)
def send_value(self, channel_value):
self.channel_thread.sendall("%s\n\n" % channel_value)
Usage:
流程A中:
c = Channel()
c.public_address()
流程B中:
def msg_received(msg):
print "received:", msg
c = Channel()
c.register("public_address_string_returned_in_process_A", msg_received)
流程A中:
c.send_value("HELLO")
流程B中:
received: HELLO