Python编程：tornado实现WebSocket通讯-阿里云开发者社区

Python编程：tornado实现WebSocket通讯

2022-08-22 135

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 以下实例有点像广播通讯，支持多个客户端连接代码参考别人的，稍微做了一点点修改服务器端 websocket_demo.py

以下实例有点像广播通讯，支持多个客户端连接

代码参考别人的，稍微做了一点点修改

服务器端 websocket_demo.py

# -*- coding: utf-8 -*-

from __future__ import unicode_literals, print_function
from datetime import datetime
import os

import tornado
from tornado.options import define, options
from tornado.web import RequestHandler
from tornado.websocket import WebSocketHandler

# 设置服务器端口
define("port", default=2222, type=int)


class IndexHandler(RequestHandler):
    def get(self):
        self.render("chat-client.html")


class ChatHandler(WebSocketHandler):
    users = set()  # 用来存放在线用户的容器

    def open(self):
        # 建立连接后添加用户到容器中
        self.users.add(self)

        # 向已在线用户发送消息
        for user in self.users:
            remote_ip, port = self.request.connection.context.address
            now = datetime.now().strftime("%H:%M:%S")
            user.write_message("[{}][{}:{}]-进入聊天室".format(now, remote_ip, port))

    def on_message(self, message):
        # 向在线用户广播消息
        now = datetime.now().strftime("%H:%M:%S")
        remote_ip, port = self.request.connection.context.address
        for user in self.users:
            user.write_message("[{}][{}:{}] {}".format(now, remote_ip, port, message))

    def on_close(self):
        # 用户关闭连接后从容器中移除用户
        now = datetime.now().strftime("%H:%M:%S")
        remote_ip, port = self.request.connection.context.address
        self.users.remove(self)
        for user in self.users:
            user.write_message("[{}][{}:{}]-离开聊天室".format(now, remote_ip, port))

    def check_origin(self, origin):
        return True  # 允许WebSocket的跨域请求


if __name__ == '__main__':
    tornado.options.parse_command_line()

    app = tornado.web.Application([
        (r"/", IndexHandler),
        (r"/chat", ChatHandler),
    ],
        static_path=os.path.join(os.path.dirname(__file__), "static"),
        template_path=os.path.join(os.path.dirname(__file__), "template"),
        debug=True
    )

    http_server = tornado.httpserver.HTTPServer(app)
    http_server.listen(options.port)
    tornado.ioloop.IOLoop.current().start()