基于Python和PostgreSQL的异步TCP服务器有效性、性能及测试问询

我有嵌入式C编程经验，但刚接触Python。设计了一套传感器电路，设备通过GPRS连接TCP服务器上报数据，现有1000-2000台设备7*24小时保持连接，每台每分钟发送100-150字节数据，服务器需要解析数据并写入PostgreSQL。

我选择异步方案开发可扩展TCP服务器，写了如下代码，但没法做千级设备实测，不确定代码是否真的实现了无CPU阻塞的异步运行；另外想知道如果代码符合异步预期，1核2GB内存的VPS能不能支撑上述流量。

另外我可以配置设备分时上报，把每秒数据接收量控制在约40条（2000台每分钟上报一次，2000/60≈33.3，约40条/秒），希望能解答这些疑问。

import asyncio
import asyncpg
#async olup olmadığı bilinmiyor.

class EchoServerProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport   

        
    def data_received(self, data):
        message = data.decode()
        print('Data received: {!r}'.format(message))
        
        database_handle()
        
        print('Send: {!r}'.format(message))
        self.transport.write(data)

        print('Close the client socket')
        self.transport.close()

    def connection_lost(self, transport):
        peername = self.transport.get_extra_info('peername')
        print('{} is disconnected from server'.format(peername))
        return super().connection_lost(transport)

async def database_handle(data):
    async with poolsql.acquire() as connection:
    # Open a transaction.
        async with connection.transaction():
        # Run the query passing the request argument.
            result = await connection.execute('''INSERT INTO deneme(data) VALUES($1)''',data)
            print(result)
    
async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    global poolsql
    poolsql=await asyncpg.create_pool('postgres://username:password@localhost:5432/test')
    
    loop = asyncio.get_running_loop()
    server = await loop.create_server(
        lambda: EchoServerProtocol(),
        '127.0.0.1', 8544, backlog=65000)
    
    async with server:
        await server.serve_forever()


asyncio.run(main())

1. 代码的异步缺陷 #

你的代码没有实现真正的无阻塞异步，核心问题在data_received方法里：

• database_handle()是异步函数，但直接调用未加await，会导致它仅返回协程对象而不实际执行，还会触发运行时错误，同时阻塞事件循环。
• asyncio.Protocol是基于回调的低层级API，回调本身是同步的，若在data_received中处理耗时操作（哪怕是错误调用异步函数），都会卡住整个事件循环，无法处理其他连接。

修复方案：

• 改用asyncio.start_server高层级异步API，直接在处理函数中用await调用异步数据库操作，避免回调阻塞。
• 若坚持用Protocol，需用asyncio.create_task(database_handle(message))将数据库操作包装成异步任务，交给事件循环调度。

修复后的示例（高层级API版本）：

import asyncio
import asyncpg

async def handle_client(reader, writer):
    peername = writer.get_extra_info('peername')
    print(f'Connection from {peername}')
    
    data = await reader.read(200)  # 根据实际数据包大小调整
    message = data.decode()
    print(f'Data received: {message!r}')
    
    # 异步执行数据库操作，不阻塞事件循环
    await database_handle(message)
    
    writer.write(data)
    await writer.drain()
    
    print(f'Close the client socket for {peername}')
    writer.close()
    await writer.wait_closed()

async def database_handle(data):
    async with poolsql.acquire() as connection:
        async with connection.transaction():
            result = await connection.execute('''INSERT INTO deneme(data) VALUES($1)''', data)
            print(result)

async def main():
    global poolsql
    poolsql = await asyncpg.create_pool('postgres://username:password@localhost:5432/test', min_size=10, max_size=50)
    
    server = await asyncio.start_server(
        handle_client, '127.0.0.1', 8544, backlog=65000)
    
    async with server:
        await server.serve_forever()

asyncio.run(main())

2. 1核2GB VPS的性能支撑能力 #

如果代码修复为正确的异步实现，1核2GB内存的VPS完全能支撑你的流量：

• 网络流量：40条/秒，每条150字节，每秒总数据量约6KB，加上TCP头部开销也远低于1核VPS的网络带宽上限。
• 数据库压力：40次简单插入/秒，PostgreSQL在1核CPU下轻松处理（单实例每秒可承载数千次简单插入），配合asyncpg连接池（建议设置10-50的连接数），不会出现瓶颈。
• 内存占用：2GB内存足够覆盖asyncio事件循环（内存消耗极低）、PostgreSQL连接池（单连接约几MB），以及操作系统和必要进程的开销。

额外优化建议：

• 关闭代码中的print语句（或替换为异步日志系统），避免同步IO操作阻塞事件循环。
• 可开启PostgreSQL的pg_stat_statements插件监控查询性能，确保插入语句无瓶颈。
• 若后续流量增长，可考虑批量插入（攒100条数据再批量写入），进一步降低数据库压力。

内容的提问来源于stack exchange，提问作者sacredserenity