如何在两台AWS EC2实例间实现低延迟的数据双向交互？

针对两台EC2上a.py和b.py的循环依赖数据交互需求，以下是几种低延迟实现方案，按延迟从低到高排序：

1. 直接TCP套接字通信（最低延迟） #

通过点对点TCP连接实现双向数据传输，跳过中间件，延迟可达亚毫秒级（同可用区内）。适合对延迟敏感的场景，需手动处理连接可靠性。

实现要点： #

• 两台EC2部署在同一VPC内，使用私有IP通信，规避公网路由延迟。
• 开放EC2安全组对应端口（如65432、65433），允许彼此私有IP访问。
• 启用TCP_NODELAY禁用Nagle算法，减少小数据包的延迟堆积。

a.py 示例代码：

import socket

# b.py所在EC2的私有IP和监听端口
B_PRIVATE_IP = '10.0.0.2'
B_PORT = 65432

def handle_interaction():
    while True:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            try:
                s.connect((B_PRIVATE_IP, B_PORT))
                # 接收b.py生成的数据
                raw_data = s.recv(4096).decode()
                if not raw_data:
                    continue
                
                # 业务处理：生成供b.py使用的数据
                processed_data = f"a_processed:{raw_data}"
                
                # 发送处理后的数据给b.py
                s.sendall(processed_data.encode())
                
                # 检查终止条件（替换为你的业务逻辑）
                if processed_data.endswith("terminate"):
                    print("终止条件触发，退出循环")
                    break
            except ConnectionRefusedError:
                # 连接失败时短暂重试
                continue

if __name__ == "__main__":
    handle_interaction()

b.py 示例代码：

import socket
import time

# a.py所在EC2的私有IP
A_PRIVATE_IP = '10.0.0.1'
# 自身监听端口
B_LISTEN_PORT = 65432

def handle_interaction():
    # 启动TCP监听，等待a.py连接
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        s.bind(('', B_LISTEN_PORT))
        s.listen()
        conn, addr = s.accept()
        with conn:
            print(f"与a.py建立连接：{addr}")
            while True:
                # 业务处理：生成供a.py使用的数据
                raw_data = f"b_generated:{time.time()}"
                
                # 发送数据给a.py
                conn.sendall(raw_data.encode())
                
                # 接收a.py处理后的数据
                processed_data = conn.recv(4096).decode()
                if not processed_data:
                    break
                
                # 检查终止条件
                if processed_data.endswith("terminate"):
                    print("终止条件触发，退出循环")
                    break

if __name__ == "__main__":
    handle_interaction()

2. AWS SQS FIFO队列（可靠低延迟） #

如果需要数据可靠传递（避免丢包），可使用SQS FIFO队列，延迟在几十毫秒级（同可用区内）。通过长轮询减少空轮询开销。

实现要点： #

• 创建两个FIFO队列（a-queue.fifo、b-queue.fifo），开启内容去重和分组功能。
• 使用长轮询（WaitTimeSeconds=20），有消息立即返回，无消息则等待最多20秒，降低延迟。

a.py 示例代码：

import boto3

sqs = boto3.client('sqs', region_name='us-east-1')
B_QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/b-queue.fifo'
A_QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/a-queue.fifo'

def handle_interaction():
    while True:
        # 长轮询接收b.py的消息
        response = sqs.receive_message(
            QueueUrl=B_QUEUE_URL,
            WaitTimeSeconds=20,
            MaxNumberOfMessages=1,
            MessageGroupId='inter-group'
        )
        
        if 'Messages' in response:
            for msg in response['Messages']:
                raw_data = msg['Body']
                # 业务处理
                processed_data = f"a_processed:{raw_data}"
                
                # 发送到a队列供b.py消费
                sqs.send_message(
                    QueueUrl=A_QUEUE_URL,
                    MessageBody=processed_data,
                    MessageGroupId='inter-group',
                    MessageDeduplicationId=f"a-{raw_data}"
                )
                
                # 删除已处理的消息
                sqs.delete_message(
                    QueueUrl=B_QUEUE_URL,
                    ReceiptHandle=msg['ReceiptHandle']
                )
                
                if processed_data.endswith("terminate"):
                    return

if __name__ == "__main__":
    handle_interaction()

关键优化建议 #

• 网络环境：将两台EC2部署在同一个可用区（AZ），并加入集群放置组，实例间网络延迟可降至亚毫秒级。
• 连接复用：TCP方案中使用长连接，避免每次交互新建连接的三次握手开销。
• 队列配置：SQS FIFO队列启用长轮询，关闭短轮询的空请求延迟。

内容的提问来源于stack exchange，提问作者sam.hay