使用channels.generic.websocket在Python中实现实时投票功能

发布时间：2023-12-26 18:56:28

在Python中，可以使用Django Channels框架的channels.generic.websocket模块来实现实时投票功能。下面是一个简单的实现示例：

1. 首先，确保已经安装了channels库。可以使用以下命令安装：

pip install channels

2. 创建Django项目，并在项目下的settings.py中加入channels应用程序：

INSTALLED_APPS = [
    ...
    'channels',
]

3. 创建一个名为vote的应用程序，并在apps.py文件中注册应用程序：

from django.apps import AppConfig

class VoteConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'vote'

4. 在vote应用程序中创建一个名为consumers.py的文件，并定义一个WebSocket消费者类：

from channels.generic.websocket import AsyncWebsocketConsumer
import json

class VoteConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = 'vote_%s' % self.room_name

        # 加入投票房间组
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )

        await self.accept()

    async def disconnect(self, close_code):
        # 离开投票房间组
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )

    # 从WebSocket接收投票数据
    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        vote = text_data_json['vote']

        # 发送投票结果到投票房间组
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'vote_message',
                'vote': vote
            }
        )

    # 从房间组接收投票结果
    async def vote_message(self, event):
        vote = event['vote']

        # 发送投票结果到WebSocket
        await self.send(text_data=json.dumps({
            'vote': vote
        }))

5. 在vote应用程序中创建一个名为routing.py的文件，并定义一个路由器：

from django.urls import re_path

from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/vote/(?P<room_name>\w+)/$', consumers.VoteConsumer.as_asgi()),
]

6. 在Django项目的根目录下的asgi.py中修改应用程序：

from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application

import vote.routing

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": URLRouter(
        vote.routing.websocket_urlpatterns
    ),
})

7. 创建一个名为vote.html的HTML模板文件，用于前端投票界面：

<!DOCTYPE html>
<html>
<head>
    <title>实时投票</title>
    <script src="https://code.jquery.com/jquery-3.6.0.min.js"></script>
</head>
<body>
    <h1>实时投票</h1>

    <div id="voteForm">
        <input type="radio" name="vote" value="option1">选项1<br>
        <input type="radio" name="vote" value="option2">选项2<br>
        <input type="radio" name="vote" value="option3">选项3<br>
        <button type="button" onclick="vote()">投票</button>
    </div>

    <div id="voteResult"></div>

    <script>
        var voteSocket = new WebSocket(
            'ws://' + window.location.host + '/ws/vote/' + roomName + '/'
        );

        voteSocket.onmessage = function(e) {
            var data = JSON.parse(e.data);
            document.querySelector('#voteResult').innerHTML = '当前投票结果：' + data.vote;
        };

        function vote() {
            var selectedOption = document.querySelector('input[name="vote"]:checked');
            var vote = selectedOption.value;

            voteSocket.send(JSON.stringify({
                'vote': vote
            }));

            selectedOption.checked = false;
        }
    </script>
</body>
</html>

8. 在Django项目的根目录下的urls.py中添加以下路由：

from django.urls import path
from vote.views import VoteView

urlpatterns = [
    path('vote/<str:room_name>/', VoteView.as_view(), name='vote'),
]

9. 在views.py中编写相应的View类：

from django.views.generic.base import TemplateView

class VoteView(TemplateView):
    template_name = 'vote.html'

现在，当用户打开/vote/<room_name>/的链接时，将会看到一个简单的投票界面，并且可以实时在网页上看到投票结果的变化。用户的投票数据将通过WebSocket发送到服务器，并通过投票房间组广播给其他用户，其他用户收到广播后将数据实时更新到网页上。

需要注意的是，上述示例中的代码非完整代码，只包含关键部分，实际使用时还需要根据具体需求进行完善。