如何在Python中使用套接字创建通道

Question

我几个月前开始学习Python，现在我正在创建一个套接字服务器。我已经让服务器能够同时处理多个线程和多个客户端（太棒了！）但我想实现一些我不知道是否存在的功能。我想创建一种频道，让客户端可以发送不同类型的消息。

举个例子，我创建一个频道叫做INFO，如果服务器收到这种类型的消息，它就会打印出来。

我还创建了另一个频道叫做DEBUG，在这个频道里我可以发送自定义命令，服务器会执行这些命令。

等等。

用非编程的语言来说，就是这样：

def socketDebug(command):
     run command

def socketInfo(input):
     print input

if socket == socketDebug:
     socketDebug(socket.rcv)
else:
   if socket == socketInfo:
     socketInfo(socket.rcv)

我希望我说得够清楚。

Answer 1

这里有一个相对简单的 Channel 类的实现。它创建了一个套接字，用来接受客户端的连接和发送消息。它自己也是一个客户端，可以接收来自其他 Channel 实例的消息（比如在不同的进程中）。

通信是通过两个线程来完成的，这样其实不是很好（我会使用异步输入输出）。当收到消息时，它会在接收线程中调用注册的函数，这可能会引发一些线程问题。

每个 Channel 实例都会创建自己的套接字，但如果由一个实例来处理多个“频道主题”，那样会更具扩展性。

一些现有的库提供了“频道”功能，比如 nanomsg。

这里的代码是为了教育目的，如果能帮到你就好了……

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)

使用方法：

在进程 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