— 1. TCP连接建立
— 客户端连接到服务器端口（默认5432）
— 三次握手建立TCP连接

— 2. 启动消息（StartupMessage）
— 客户端发送启动消息，包含协议版本和参数

— 启动消息格式：
— Int32 – 消息长度（包括自身）
— Int32 – 协议版本（196608表示3.0）
— String – 参数名
— String – 参数值
— … (更多参数)

— 示例启动消息参数：
— user: fgedu
— database: fgedudb
— client_encoding: UTF8
— DateStyle: ISO, MDY

— 3. 认证交换
— 服务器发送认证请求消息

— AuthenticationOk (R, 0)
— AuthenticationCleartextPassword (R, 3)
— AuthenticationMD5Password (R, 5)
— AuthenticationSASL (R, 10)

— 客户端响应密码消息
— PasswordMessage (p)

— 4. 参数状态
— 服务器发送参数状态消息

— ParameterStatus (S)
— 参数名: server_version
— 参数值: 18.0

— ParameterStatus (S)
— 参数名: server_encoding
— 参数值: UTF8

— 5. 后端密钥数据
— BackendKeyData (K)
— 进程ID: 12345
— 密钥: 67890

— 6. 就绪查询
— ReadyForQuery (Z)
— 状态: I (空闲), T (事务中), E (失败)

— 使用psql查看连接信息
SELECT * FROM pg_stat_activity WHERE pid = pg_backend_pid();

— 输出结果
pid | query | state | backend_start
——-+————————-+——–+———————
12345 | SELECT * FROM pg_stat…| active | 2026-04-07 16:00:00
(1 row)

— 查看当前连接参数
SHOW ALL;

— 输出结果（部分）
name | setting
—————————-+—————————————–
server_version | 18.0
server_encoding | UTF8
client_encoding | UTF8
DateStyle | ISO, MDY
TimeZone | Asia/Shanghai
(5 rows)

— 1. SSL协商
— 客户端发送SSLRequest消息
— Int32(8) – 消息长度
— Int32(80877103) – SSL请求码

— 2. 服务器响应
— ‘S’ – 支持SSL，开始SSL握手
— ‘N’ – 不支持SSL，继续普通连接

— 3. SSL握手
— 标准TLS/SSL握手过程

— 配置SSL连接
— postgresql.conf
ssl = on
ssl_cert_file = ‘/postgresql/ssl/server.crt’
ssl_key_file = ‘/postgresql/ssl/server.key’
ssl_ca_file = ‘/postgresql/ssl/ca.crt’

— pg_hba.conf
hostssl all all 192.168.1.0/24 md5

— 重新加载配置
SELECT pg_reload_conf();

— 输出结果
pg_reload_conf
—————-
t
(1 row)

— 查看SSL连接状态
SELECT ssl, version, cipher
FROM pg_stat_ssl
WHERE pid = pg_backend_pid();

— 输出结果
ssl | version | cipher
—–+———+———————————-
t | TLSv1.3 | TLS_AES_256_GCM_SHA384
(1 row)

— 强制SSL连接测试
$ psql “host=192.168.1.100 dbname=fgedudb user=fgedu sslmode=require”

— 输出结果
psql (18.0)
SSL connection (protocol: TLSv1.3, cipher: TLS_AES_256_GCM_SHA384)
Type “help” for help.

fgedudb=#

— 1. 客户端发送Query消息
— Query (Q)
— String – SQL查询语句

— 2. 服务器响应
— RowDescription (T) – 行描述
— DataRow (D) – 数据行（多条）
— CommandComplete (C) – 命令完成
— ReadyForQuery (Z) – 就绪

— 简单查询示例
SELECT id, customer_name, phone FROM fgedu_customers LIMIT 3;

— 输出结果
id | customer_name | phone
—-+—————+————–
1 | 张三 | 13800138000
2 | 李四 | 13900139000
3 | 王五 | 13700137000
(3 rows)

— 消息序列（简化）：
— 前端 -> 后端: Q “SELECT id, customer_name, phone FROM fgedu_customers LIMIT 3”
— 后端 -> 前端: T (RowDescription)
— – 字段1: id (int4)
— – 字段2: customer_name (varchar)
— – 字段3: phone (varchar)
— 后端 -> 前端: D (DataRow) [1, “张三”, “13800138000”]
— 后端 -> 前端: D (DataRow) [2, “李四”, “13900139000”]
— 后端 -> 前端: D (DataRow) [3, “王五”, “13700137000”]
— 后端 -> 前端: C “SELECT 3”
— 后端 -> 前端: Z ‘I’

— 错误响应示例
SELECT * FROM nonexistent_table;

— 输出结果
ERROR: relation “nonexistent_table” does not exist
LINE 1: SELECT * FROM nonexistent_table;

— 错误消息序列：
— 后端 -> 前端: E (ErrorResponse)
— – S: ERROR
— – C: 42P01
— – M: relation “nonexistent_table” does not exist
— – L: 1
— 后端 -> 前端: Z ‘I’

— 使用pgbench测试查询性能
$ pgbench -c 10 -t 1000 -S fgedudb

— 输出结果
transaction type:
scaling factor: 1
query mode: simple
number of clients: 10
number of threads: 1
maximum number of tries: 1
number of transactions per client: 1000
number of transactions actually processed: 10000/10000
number of failed transactions: 0 (0.000%)
latency average = 0.500 ms
initial connection time = 50.123 ms
tps = 20000.000000 (without initial connection time)

— 扩展查询模式将查询分为多个阶段：
— Parse -> Bind -> Execute -> Sync

— 1. Parse消息
— Parse (P)
— String – 语句名称（空字符串表示匿名语句）
— String – 查询语句
— Int16 – 参数数量
— Int32[] – 参数类型OID

— 2. Bind消息
— Bind (B)
— String – portal名称
— String – 语句名称
— Int16 – 参数格式代码数量
— Int16[] – 参数格式代码
— Int16 – 参数值数量
— Int32[] – 参数值长度
— Byte[] – 参数值
— Int16 – 结果格式代码数量
— Int16[] – 结果格式代码

— 3. Describe消息
— Describe (D)
— Byte – ‘S’ (语句) 或 ‘P’ (portal)
— String – 语句或portal名称

— 4. Execute消息
— Execute (E)
— String – portal名称
— Int32 – 最大返回行数（0表示无限制）

— 5. Sync消息
— Sync (S)
— 标记扩展查询边界

— 使用PREPARE模拟扩展查询
PREPARE fgedu_get_customer(int) AS
SELECT customer_name, phone FROM fgedu_customers WHERE id = $1;

— 输出结果
PREPARE

— 执行预处理语句
EXECUTE fgedu_get_customer(1);

— 输出结果
customer_name | phone
—————+————–
张三 | 13800138000
(1 row)

— 查看预处理语句
SELECT name, statement, parameter_types
FROM pg_prepared_statements
WHERE name = ‘fgedu_get_customer’;

— 输出结果
name | statement | parameter_types
———————+————————————————–+—————–
fgedu_get_customer | PREPARE fgedu_get_customer(int) AS | {integer}
| SELECT customer_name, phone FROM fgedu_customers |
| WHERE id = $1 |
(1 row)

— 批量执行示例
PREPARE fgedu_insert_order(varchar, int, numeric) AS
INSERT INTO fgedu_orders(order_no, customer_id, amount)
VALUES($1, $2, $3);

— 输出结果
PREPARE

— 批量执行
EXECUTE fgedu_insert_order(‘ORD101’, 1, 1000.00);
EXECUTE fgedu_insert_order(‘ORD102’, 2, 2000.00);
EXECUTE fgedu_insert_order(‘ORD103’, 3, 3000.00);

— 输出结果
INSERT 0 1
INSERT 0 1
INSERT 0 1

— 清理预处理语句
DEALLOCATE fgedu_get_customer;
DEALLOCATE fgedu_insert_order;

— 输出结果
DEALLOCATE
DEALLOCATE

— 1. 客户端发送CopyData消息
— CopyData (d)
— Byte[n] – 数据

— 2. 服务器响应
— CopyInResponse (G) – 开始COPY IN
— CopyData (d) – 数据（多条）
— CopyDone (c) – 结束COPY
— CommandComplete (C) – 命令完成

— COPY IN示例
COPY fgedu_customers(customer_name, phone, email)
FROM STDIN
WITH (FORMAT CSV, HEADER true, DELIMITER ‘,’);

— 输入数据
张三,13800138000,zhangsan@fgedu.net.cn
李四,13900139000,lisi@fgedu.net.cn
王五,13700137000,wangwu@fgedu.net.cn
\.

— 输出结果
COPY 3

— 使用COPY导入文件
COPY fgedu_products
FROM ‘/postgresql/data/products.csv’
WITH (FORMAT CSV, HEADER true);

— 输出结果
COPY 100

— 查看导入结果
SELECT COUNT(*) FROM fgedu_products;

— 输出结果
count
——-
100
(1 row)

— COPY OUT流程（服务器到客户端）
COPY (SELECT * FROM fgedu_orders WHERE create_time >= ‘2026-04-01’)
TO STDOUT
WITH (FORMAT CSV, HEADER true);

— 输出结果
id,order_no,customer_id,amount,status,create_time
1,ORD001,1,1000.00,pending,2026-04-01 10:00:00
2,ORD002,2,2000.00,paid,2026-04-02 11:00:00
…

— 导出到文件
COPY fgedu_orders
TO ‘/postgresql/data/orders_export.csv’
WITH (FORMAT CSV, HEADER true);

— 输出结果
COPY 1000

— 查看导出文件
$ head -5 /postgresql/data/orders_export.csv

— 输出结果
id,order_no,customer_id,amount,status,create_time
1,ORD001,1,1000.00,pending,2026-04-01 10:00:00
2,ORD002,2,2000.00,paid,2026-04-02 11:00:00
3,ORD003,3,3000.00,shipped,2026-04-03 12:00:00
4,ORD004,1,4000.00,pending,2026-04-04 13:00:00

— 使用tcpdump捕获PostgreSQL流量
$ tcpdump -i eth0 -s 0 -w pg_traffic.pcap port 5432

— 输出结果
tcpdump: listening on eth0, link-type EN10MB (Ethernet), capture size 262144 bytes
^C
100 packets captured
200 packets received by filter
0 packets dropped by kernel

— 使用Wireshark分析
— 打开pg_traffic.pcap文件
— 过滤器: tcp.port == 5432

— 使用pg_recvlogical分析逻辑复制协议
$ pg_recvlogical -h 192.168.1.100 -U fgedu -d fgedudb \
–slot=test_slot –start -f –

— 输出结果
BEGIN 1234
table public.fgedu_orders: INSERT: id[integer]:1 order_no[character varying]:’ORD001′
COMMIT 1234

— 启用协议调试日志
— postgresql.conf
log_connections = on
log_disconnections = on
log_statement = ‘all’

— 重新加载配置
SELECT pg_reload_conf();

— 输出结果
pg_reload_conf
—————-
t
(1 row)

— 查看连接日志
$ tail -f /postgresql/fgedata/log/postgresql-2026-04-07.log

— 输出结果
2026-04-07 16:00:00.123 CST [12345] LOG: connection received: host=192.168.1.100 port=54321
2026-04-07 16:00:00.456 CST [12345] LOG: connection authorized: user=fgedu database=fgedudb
2026-04-07 16:00:01.789 CST [12345] LOG: statement: SELECT * FROM fgedu_customers;
2026-04-07 16:00:02.123 CST [12345] LOG: disconnection: session time: 0:00:02.000 user=fgedu database=fgedudb host=192.168.1.100

— 使用pg_stat_statements分析查询
CREATE EXTENSION pg_stat_statements;

— 输出结果
CREATE EXTENSION

SELECT query, calls, total_time, mean_time, rows
FROM pg_stat_statements
ORDER BY total_time DESC
LIMIT 10;

— 输出结果
query | calls | total_time | mean_time | rows
——————————————————+——-+————+———–+——-
SELECT * FROM fgedu_orders WHERE customer_id = $1 | 1000 | 5000.00 | 5.00 | 50000
SELECT * FROM fgedu_customers WHERE id = $1 | 500 | 2000.00 | 4.00 | 500
INSERT INTO fgedu_orders VALUES(…) | 200 | 1000.00 | 5.00 | 200
(3 rows)

— Python客户端示例
— fgedu_pg_client.py

#!/usr/bin/env python3
# fgedu_pg_client.py
# from:www.itpux.com.qq113257174.wx:itpux-com
# web: http://www.fgedu.net.cn

import socket
import struct
import hashlib

class FgeduPgClient:
def __init__(self, host, port=5432):
self.host = host
self.port = port
self.sock = None

def connect(self, user, database, password=None):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect((self.host, self.port))

# 发送启动消息
startup_msg = self._build_startup_message(user, database)
self.sock.send(startup_msg)

# 处理认证
while True:
msg_type, msg_data = self._read_message()

if msg_type == b’R’: # Authentication
auth_type = struct.unpack(‘!I’, msg_data[:4])[0]
if auth_type == 0: # AuthenticationOk
continue
elif auth_type == 3: # CleartextPassword
self._send_password(password)
elif auth_type == 5: # MD5Password
salt = msg_data[4:8]
self._send_md5_password(user, password, salt)
elif msg_type == b’S’: # ParameterStatus
pass
elif msg_type == b’K’: # BackendKeyData
pass
elif msg_type == b’Z’: # ReadyForQuery
break

def _build_startup_message(self, user, database):
params = {
‘user’: user,
‘database’: database,
‘client_encoding’: ‘UTF8’
}

msg = struct.pack(‘!I’, 196608) # Protocol 3.0
for key, value in params.items():
msg += key.encode() + b’\x00′ + value.encode() + b’\x00′
msg += b’\x00′

length = len(msg) + 4
return struct.pack(‘!I’, length) + msg

def _send_password(self, password):
msg = password.encode() + b’\x00′
length = len(msg) + 4
self.sock.send(b’p’ + struct.pack(‘!I’, length) + msg)

def _send_md5_password(self, user, password, salt):
if password is None:
password = ”
hash1 = hashlib.md5((password + user).encode()).hexdigest()
hash2 = hashlib.md5((hash1 + salt.decode(‘latin1’)).encode()).hexdigest()
msg = (‘md5′ + hash2).encode() + b’\x00′
length = len(msg) + 4
self.sock.send(b’p’ + struct.pack(‘!I’, length) + msg)

def _read_message(self):
msg_type = self.sock.recv(1)
length = struct.unpack(‘!I’, self.sock.recv(4))[0]
msg_data = self.sock.recv(length – 4)
return msg_type, msg_data

def query(self, sql):
# 发送Query消息
msg = sql.encode() + b’\x00′
length = len(msg) + 4
self.sock.send(b’Q’ + struct.pack(‘!I’, length) + msg)

# 读取响应
results = []
while True:
msg_type, msg_data = self._read_message()

if msg_type == b’T’: # RowDescription
pass
elif msg_type == b’D’: # DataRow
results.fgappend(msg_data)
elif msg_type == b’C’: # CommandComplete
pass
elif msg_type == b’Z’: # ReadyForQuery
break

return results

def close(self):
if self.sock:
self.sock.close()

# 使用示例
if __name__ == ‘__main__’:
client = FgeduPgClient(‘192.168.1.100’)
client.connect(‘fgedu’, ‘fgedudb’, ‘fgedu_2026’)
results = client.query(‘SELECT * FROM fgedu_customers LIMIT 5’)
print(f”Received {len(results)} rows”)
client.close()

— 运行客户端
$ python3 fgedu_pg_client.py

— 输出结果
Received 5 rows

— 启用调试日志
— postgresql.conf
client_min_messages = debug5
log_min_messages = debug5
log_error_verbosity = verbose

— 重新加载配置
SELECT pg_reload_conf();

— 输出结果
pg_reload_conf
—————-
t
(1 row)

— 使用strace跟踪系统调用
$ strace -e trace=network -p 12345

— 输出结果
Process 12345 attached
recvfrom(7, “Q\0\0\0\x1eSELECT * FROM fgedu_customers”…, 8192, 0, NULL, NULL) = 34
sendto(7, “T\0\0\0\x44\0\3id\0\0\0\0\0\0\0\0\0\0\0″…, 8192, 0, NULL, 0) = 123

— 使用gdb调试PostgreSQL进程
$ gdb -p 12345

— 输出结果
(gdb) break pq_getbyte
Breakpoint 1 at 0x123456: file pqcomm.c, line 1000.
(gdb) continue
Continuing.

— 查看当前连接状态
SELECT
pid,
state,
query,
wait_event_type,
wait_event
FROM pg_stat_activity
WHERE pid = pg_backend_pid();

— 输出结果
pid | state | query | wait_event_type | wait_event
——-+——–+————————-+—————–+————
12345 | active | SELECT * FROM pg_stat…| Client | ClientRead
(1 row)

— 查看网络连接信息
SELECT
pid,
client_addr,
client_port,
backend_start,
state
FROM pg_stat_activity
WHERE client_addr IS NOT NULL
ORDER BY backend_start DESC;

— 输出结果
pid | client_addr | client_port | backend_start | state
——-+—————+————-+———————–+——–
12345 | 192.168.1.100 | 54321 | 2026-04-07 16:00:00 | active
12346 | 192.168.1.101 | 54322 | 2026-04-07 15:30:00 | idle
(2 rows)

— 查看协议版本
SELECT setting FROM pg_settings WHERE name = ‘server_version’;

— 输出结果
setting
———
18.0
(1 row)

— 测试连接超时
— postgresql.conf
statement_timeout = 30000
lock_timeout = 10000

— 测试查询超时
SET statement_timeout = 1000;
SELECT pg_sleep(2);

— 输出结果
ERROR: canceling statement due to statement timeout

— 查看取消请求
SELECT * FROM pg_stat_activity WHERE state = ‘active’;

— 输出结果
pid | state | query | wait_event_type | wait_event
——-+——–+—————–+—————–+————
12345 | active | SELECT pg_sleep | Timeout | Timeout
(1 row)