#UDP Programming

#Welcome to UDP efficient communication practice

Why does the video call freeze for a few seconds and then jump directly? Why can DNS resolution achieve millisecond response? Why doesn’t your LAN gaming team have to wait for the handshake prompt? Today we will dismantle the "lightweight player" behind it - UDP protocol. From basic API to modern best practices, we will use Python to take you into practice.

#1. Understand UDP first: it’s not “unreliable”, it’s “choose as needed”

UDP (User Datagram Protocol) is a connectionless protocol of the transport layer. Compared with the TCP we are more familiar with, it leaves all the complex reliability, sequence, and flow control mechanisms to the application layer to decide whether to add them - in exchange for ultimate lightness and speed.

#One-sentence summary of core features

• No connection: Instead of shaking hands three times and waving four times, just insert the address before sending the message, just like a courier knocking on the door and inserting a package (it doesn’t matter whether you are at home or not, and whether you are willing to accept it or not);
• Unreliable and not sorted: There is no guarantee that the package will be delivered, the order is not guaranteed, and it does not even tell you that it is lost/disordered;
• Extremely lightweight: The header is only 8 bytes (the TCP header is at least 20 bytes and at most 60 bytes);
• Support broadcast/multicast: Can send "bulk broadcast" to all devices in the same LAN/specified group.

#When should I use UDP?

✅ Real-time priority, high packet loss error: Video streaming, voice calls, live barrages (losing 1 frame/barrage does not affect the overall experience); ✅ Query small data: DNS, DHCP (a few KB per interaction, just resend after timeout, using TCP is a waste of time); ✅ LAN/Multicast scenario: Device discovery, game team notification.

#2. Introduction to Python UDP: Write a small sending and receiving program in 10 minutes

PythonsocketThe library natively supports UDP and does not require additional installation dependencies. Let's first write the most basic "one send and one receive" version of the server and client.

#Basic version server

Bind the local IP and port, receive data in an infinite loop, and return the response in the same way after receiving it (note the use ofrecvfrom()Getting the data + client address at the same time is the core of UDP programming! ).

import socket

# 1. 创建UDP专用socket：AF_INET是IPv4，SOCK_DGRAM是UDP
server_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 2. 绑定服务器地址（0.0.0.0表示监听所有网卡，这里先用127.0.0.1本地测试）
server_addr = ("127.0.0.1", 9999)
server_sock.bind(server_addr)
print(f"✅ UDP本地服务器已启动，监听地址：{server_addr[0]}:{server_addr[1]}")

try:
    while True:
        print("\n⏳ 等待接收客户端消息...")
        # 3. 接收数据：每次最多收1024字节，返回(数据二进制, 发送方元组)
        data, client_addr = server_sock.recvfrom(1024)
        
        print(f"📥 来自 {client_addr} 的消息：{data.decode('utf-8')}")
        
        # 4. 原路返回响应
        reply = f"📤 服务器收到了：{data.decode('utf-8')}"
        server_sock.sendto(reply.encode("utf-8"), client_addr)

except KeyboardInterrupt:
    print("\n🛑 用户手动停止服务器")
finally:
    server_sock.close()
    print("🔌 服务器socket已关闭")

#Basic version client

There is no need to bind an address, just send a message directly to the specified server and wait for the response.

import socket

# 1. 创建客户端socket（可以复用同一个，也不用bind）
client_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 指定服务器地址
server_addr = ("127.0.0.1", 9999)

# 测试消息列表
test_msgs = ["你好呀UDP！", "Python编程真有意思", "试试最后一条消息"]

try:
    for msg in test_msgs:
        print(f"📤 发送消息：{msg}")
        client_sock.sendto(msg.encode("utf-8"), server_addr)
        
        # 2. 接收响应（同样限制1024字节）
        reply, _ = client_sock.recvfrom(1024)  # 这里服务器地址可以用_忽略
        print(f"📥 收到回复：{reply.decode('utf-8')}\n")

except Exception as e:
    print(f"❌ 出错了：{e}")
finally:
    client_sock.close()
    print("🔌 客户端socket已关闭")

#3. Modern UDP best practices: Don’t just write the entry version!

The entry version only runs through the process. The real production/development environment needs to solve problems such as timeouts, exceptions, data security, and high concurrency. Here is an optimized version of JSON server for you.

#Read optimization points in advance

• with statement automatically closes socket: no need to manually try/finally close it to avoid resource leakage;
• Timeout Setting: Avoid infinite waiting for the server/client;
• Exception layered processing: distinguish JSON parsing errors, network timeouts, and other unknown errors;
• JSON Serialization: Structured transmission of data is much more convenient than plain text.

#Optimized JSON server code

import socket
import json

def run_json_udp_server():
    # 用with语句自动管理socket生命周期
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.bind(("0.0.0.0", 9999))
        sock.settimeout(5.0)  # 5秒没收到数据就触发超时异常
        print("✅ 优化版JSON UDP服务器已启动，支持结构化数据！")

        while True:
            try:
                data, client_addr = sock.recvfrom(1024)
                try:
                    # 尝试解析JSON
                    req_msg = json.loads(data.decode("utf-8"))
                    print(f"📥 来自 {client_addr} 的结构化请求：{req_msg}")

                    # 构造标准化响应
                    resp_msg = {"code": 200, "status": "success", "received_data": req_msg}
                    sock.sendto(json.dumps(resp_msg).encode("utf-8"), client_addr)

                except json.JSONDecodeError:
                    # JSON格式不对的错误
                    print(f"⚠️ 来自 {client_addr} 的消息不是有效JSON")
                    error_resp = {"code": 400, "status": "error", "msg": "Invalid JSON format"}
                    sock.sendto(json.dumps(error_resp).encode("utf-8"), client_addr)

            except socket.timeout:
                # 超时只打印日志，继续监听
                print("⏸️ 5秒无消息，继续监听...")
            except KeyboardInterrupt:
                # 用户Ctrl+C停止
                print("\n🛑 用户手动停止服务器")
                break
            except Exception as e:
                # 其他未知错误，直接退出
                print(f"❌ 服务器遇到未知错误：{e}")
                break

if __name__ == "__main__":
    run_json_udp_server()

#4. UDP vs TCP: Complete understanding with one table

#5. Common UDP misunderstandings & tips

#Q1: Is UDP really "completely unusable in scenarios that require reliability?"

no! The QUIC protocol (the bottom layer of HTTP/3) achieves the reliability of all TCP + the multiplexing of HTTP/2 on the basis of UDP - it is faster than TCP and also solves the head-of-line blocking problem of TCP.

#Q2: How to improve the security of UDP program?

1. Add data encryption: Use DTLS (UDP version of TLS);
2. Add signature verification: Use HMAC to prevent data from being tampered with;
3. Limit buffer size: To prevent attackers from sending oversized packets and causing buffer overflow;
4. Add whitelist: Only allow access to specified IP/network segments.

#Q3: Is UDP suitable for transferring large files?

It is not recommended to upload directly! If you must pass it, add it yourself:

1. Fragmentation and reassembly mechanism (due to MTU limitation, a single UDP packet can transmit up to about 1472 bytes);
2. Serial number (guaranteed sequence);
3. Confirm retransmission mechanism (similar to TCP’s ACK).

#6. Summary

UDP is not an "unreliable protocol", but a lightweight tool that can be chosen on demand - handing over complex choices to the application layer, in exchange for the ultimate improvement in real-time performance and efficiency.

Today we learned:

1. Core features and applicable scenarios of UDP;
2. The basic version of the Python UDP program that sends and receives;
3. Produce a usable optimized version of JSON UDP server;
4. UDP vs TCP comparison table;
5. Common UDP misunderstandings and tips.

The next article can talk about UDP broadcast/multicast programming, or asyncio asynchronous UDP server. If you are interested, you can leave a message in the comment area~