Channels ，一个强大的 Python 库_channels python

作者：AllinToyou | 2024-03-26 11:51:47

踩

channels python

介绍

Channels 是一个Django项目的扩展，可以使开发者轻松地处理WebSocket、长轮询、HTTP2和其他非同步协议。Channels的设计目的是为了扩展Django的功能，允许它支持类似即时消息这样的特性，同时仍能保留其同步、便捷的特性。它提供了一种异步处理的方式，同时也支持完全的异步编程。

安装方式

要开始使用Django Channels，你需要安装channels库以及channels_redis，后者是作为Channels的后端来使用的。你可以使用pip来安装：

pip install channels  
pip install channels_redis  

1
2
3

在安装了channels之后，你还需要将其添加到Django项目的INSTALLED_APPS设置中：

# settings.py  
  
INSTALLED_APPS = [  
    # ...  
    'channels',  
    # ...  
]  

1
2
3
4
5
6
7
8

还需要指定CHANNEL_LAYERS来配置Channels层。如果你使用Redis作为后端，配置可能如下所示：

# settings.py  
  
CHANNEL_LAYERS = {  
    'default': {  
        'BACKEND': 'channels_redis.core.RedisChannelLayer',  
        'CONFIG': {  
            "hosts": [('127.0.0.1', 6379)],  
        },  
    },  
}  

1
2
3
4
5
6
7
8
9
10
11

使用方式

要使用Channels, 你需创建一个routing.py文件在你的Django app目录下，此文件包含了WebSocket路由的配置。你需要创建一个消费者类（Consumer class），该类定义了WebSocket连接的处理逻辑：

# myapp/consumers.py  
  
from channels.generic.websocket import AsyncWebsocketConsumer  
  
class EchoConsumer(AsyncWebsocketConsumer):  
    async def connect(self):  
        await self.accept()  
  
    async def disconnect(self, close_code):  
        pass  
  
    async def receive(self, text_data):  
        await self.send(text_data=text_data)  

1
2
3
4
5
6
7
8
9
10
11
12
13
14

然后在routing.py中引入这个消费者并设置路由：

# myapp/routing.py  
  
from django.urls import path  
from .consumers import EchoConsumer  
  
websocket_urlpatterns = [  
    path('ws/echo/', EchoConsumer.as_asgi()),  
]  

1
2
3
4
5
6
7
8
9

最后，你需要将这些WebSocket路由添加到项目的主路由配置中：

# mysite/routing.py  
  
from channels.routing import ProtocolTypeRouter, URLRouter  
import myapp.routing  
  
application = ProtocolTypeRouter({  
    "websocket": URLRouter(  
        myapp.routing.websocket_urlpatterns  
    ),  
})  

1
2
3
4
5
6
7
8
9
10
11

确保你的ASGI应用指向了这个mysite.routing.application。

代码示例

下面是一个更完整的示例，包含了一个简单的echo服务器，它回显客户端发送的所有消息。

# myapp/consumers.py  
import json  
from channels.generic.websocket import AsyncWebsocketConsumer  
  
class ChatConsumer(AsyncWebsocketConsumer):  
    async def connect(self):  
        self.room_name = self.scope['url_route']['kwargs']['room_name']  
        self.room_group_name = 'chat_%s' % self.room_name  
  
        # Join room group  
        await self.channel_layer.group_add(  
            self.room_group_name,  
            self.channel_name  
        )  
  
        await self.accept()  
  
    async def disconnect(self, close_code):  
        # Leave room group  
        await self.channel_layer.group_discard(  
            self.room_group_name,  
            self.channel_name  
        )  
  
    # Receive message from WebSocket  
    async def receive(self, text_data):  
        text_data_json = json.loads(text_data)  
        message = text_data_json['message']  
  
        # Send message to room group  
        await self.channel_layer.group_send(  
            self.room_group_name,  
            {  
                'type': 'chat_message',  
                'message': message  
            }  
        )  
  
    # Receive message from room group  
    async def chat_message(self, event):  
        message = event['message']  
  
        # Send message to WebSocket  
        await self.send(text_data=json.dumps({  
            'message': message  
        }))  
  
# myapp/routing.py  
from django.urls import path  
from .consumers import ChatConsumer  
  
websocket_urlpatterns = [  
    path('ws/chat/<str:room_name>/', ChatConsumer.as_asgi()),  
]  
  
# mysite/routing.py  
from channels.routing import ProtocolTypeRouter, URLRouter  
import myapp.routing  
  
application = ProtocolTypeRouter({  
    "websocket": URLRouter(  
        myapp.routing.websocket_urlpatterns  
    ),  
})  

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65