寻求Redis Pub/Sub到Postgres的简易ETL开箱即用方案

针对你的小规模、低频率数据同步需求，Kafka连接器方案确实过重，以下是几种更简易的开箱即用方案：

方案1：Python/Node.js 脚本直连实现（最推荐） #

直接用脚本语言实现端到端逻辑，无需额外中间件，代码可控且维护成本极低，完全适配你的场景。

核心逻辑： #

• 用Redis客户端订阅目标频道
• 解析<user_id>|<user_type>|<event_type>|...格式的消息，提取字段和参数
• 用Postgres客户端将解析后的数据插入目标表
• 可选：用systemd/supervisor将脚本注册为后台服务，确保持续运行

Python 示例代码： #

先安装依赖：

pip install redis psycopg2-binary

import redis
import psycopg2

# 配置信息，根据实际环境修改
REDIS_CONFIG = {"host": "localhost", "port": 6379, "decode_responses": True, "channel": "your-target-channel"}
PG_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "database": "your-db-name",
    "user": "your-db-user",
    "password": "your-db-pass",
    "table": "your-target-table"
}

def parse_message(raw_msg):
    """解析Redis消息为字典格式"""
    parts = raw_msg.split("|")
    parsed = {
        "user_id": parts[0].strip(),
        "user_type": parts[1].strip(),
        "event_type": parts[2].strip()
    }
    # 解析后续参数（假设格式为param_key=param_value）
    for part in parts[3:]:
        if "=" in part:
            key, val = part.split("=", 1)
            parsed[key.strip()] = val.strip()
    return parsed

def get_table_columns(pg_conn, table_name):
    """获取Postgres目标表的字段列表，确保插入字段匹配"""
    with pg_conn.cursor() as cur:
        cur.execute(f"SELECT column_name FROM information_schema.columns WHERE table_name = '{table_name}'")
        return [row[0] for row in cur.fetchall()]

def main():
    # 初始化Redis订阅
    r = redis.Redis(**{k: v for k, v in REDIS_CONFIG.items() if k != "channel"})
    pubsub = r.pubsub()
    pubsub.subscribe(REDIS_CONFIG["channel"])
    
    # 初始化Postgres连接
    pg_conn = psycopg2.connect(**{k: v for k, v in PG_CONFIG.items() if k != "table"})
    table_columns = get_table_columns(pg_conn, PG_CONFIG["table"])
    
    print(f"已订阅Redis频道 {REDIS_CONFIG['channel']}，等待消息...")
    try:
        for msg in pubsub.listen():
            if msg["type"] == "message":
                parsed_data = parse_message(msg["data"])
                # 过滤出目标表存在的字段
                insert_fields = [k for k in parsed_data.keys() if k in table_columns]
                if not insert_fields:
                    print("无匹配字段，跳过该消息")
                    continue
                insert_values = [parsed_data[k] for k in insert_fields]
                # 执行插入
                with pg_conn.cursor() as cur:
                    query = f"INSERT INTO {PG_CONFIG['table']} ({', '.join(insert_fields)}) VALUES ({', '.join(['%s']*len(insert_values))})"
                    cur.execute(query, insert_values)
                pg_conn.commit()
                print(f"成功插入数据: {parsed_data}")
    except KeyboardInterrupt:
        print("停止监听...")
    finally:
        pg_conn.close()
        pubsub.close()

if __name__ == "__main__":
    main()

部署建议： #

将脚本保存为redis_to_postgres.py，用systemd创建服务文件（如/etc/systemd/system/redis-etl.service）：

[Unit]
Description=Redis to Postgres ETL Service
After=network.target redis-server.service postgresql.service

[Service]
User=your-user
ExecStart=/usr/bin/python3 /path/to/redis_to_postgres.py
Restart=always

[Install]
WantedBy=multi-user.target

然后执行：

sudo systemctl daemon-reload
sudo systemctl start redis-etl
sudo systemctl enable redis-etl

方案2：Go 二进制程序（无依赖部署） #

如果不想安装Python/Node.js环境，可以用Go编写轻量二进制程序，编译后直接运行，性能更优且无额外依赖。

示例代码： #

package main

import (
	"context"
	"fmt"
	"strings"

	"github.com/go-redis/redis/v8"
	"github.com/jackc/pgx/v4"
)

// 配置信息，根据实际修改
var (
	redisAddr    = "localhost:6379"
	redisChannel = "your-target-channel"
	pgConnStr    = "postgres://user:password@localhost:5432/dbname"
	pgTableName  = "your-target-table"
)

func parseMsg(payload string) map[string]string {
	parts := strings.Split(payload, "|")
	data := make(map[string]string)
	if len(parts) >= 3 {
		data["user_id"] = strings.TrimSpace(parts[0])
		data["user_type"] = strings.TrimSpace(parts[1])
		data["event_type"] = strings.TrimSpace(parts[2])
	}
	for _, part := range parts[3:] {
		kv := strings.SplitN(part, "=", 2)
		if len(kv) == 2 {
			data[strings.TrimSpace(kv[0])] = strings.TrimSpace(kv[1])
		}
	}
	return data
}

func getPgColumns(conn *pgx.Conn, table string) ([]string, error) {
	var cols []string
	rows, err := conn.Query(context.Background(), fmt.Sprintf("SELECT column_name FROM information_schema.columns WHERE table_name = '%s'", table))
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var col string
		if err := rows.Scan(&col); err != nil {
			return nil, err
		}
		cols = append(cols, col)
	}
	return cols, nil
}

func main() {
	// Redis连接
	rdb := redis.NewClient(&redis.Options{Addr: redisAddr})
	defer rdb.Close()

	// Postgres连接
	pgConn, err := pgx.Connect(context.Background(), pgConnStr)
	if err != nil {
		fmt.Printf("Postgres连接失败: %v\n", err)
		return
	}
	defer pgConn.Close(context.Background())

	// 获取表字段
	cols, err := getPgColumns(pgConn, pgTableName)
	if err != nil {
		fmt.Printf("获取表字段失败: %v\n", err)
		return
	}

	// 订阅频道
	pubsub := rdb.Subscribe(context.Background(), redisChannel)
	defer pubsub.Close()

	fmt.Printf("已订阅Redis频道 %s，等待消息...\n", redisChannel)
	for {
		msg, err := pubsub.ReceiveMessage(context.Background())
		if err != nil {
			fmt.Printf("接收消息失败: %v\n", err)
			continue
		}

		parsed := parseMsg(msg.Payload)
		var fields []string
		var values []interface{}
		for k, v := range parsed {
			for _, col := range cols {
				if k == col {
					fields = append(fields, k)
					values = append(values, v)
					break
				}
			}
		}

		if len(fields) == 0 {
			fmt.Println("无匹配字段，跳过消息")
			continue
		}

		// 构造插入语句
		placeholders := make([]string, len(values))
		for i := range placeholders {
			placeholders[i] = fmt.Sprintf("$%d", i+1)
		}
		query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", pgTableName, strings.Join(fields, ", "), strings.Join(placeholders, ", "))

		if _, err := pgConn.Exec(context.Background(), query, values...); err != nil {
			fmt.Printf("插入数据失败: %v\n", err)
			continue
		}

		fmt.Printf("成功插入数据: %v\n", parsed)
	}
}

编译运行： #

go mod init redis-etl
go get github.com/go-redis/redis/v8 github.com/jackc/pgx/v4
go build -o redis-etl
./redis-etl

同样可以用systemd注册为后台服务。

方案3：低代码工具（无需写代码） #

如果不想编写代码，可以用N8N或Make这类低代码工作流工具，本地部署后通过可视化配置完成流程：

• 配置Redis节点：选择Redis Pub/Sub订阅模式，填入频道信息
• 配置解析节点：用Function或Set节点拆分消息格式，提取字段
• 配置Postgres节点：选择Insert模式，映射解析后的字段到目标表列
• 启动工作流，持续监听并同步数据

这类工具开箱即用，适合非开发人员或快速搭建场景。

方案对比 #

对于你的需求，Python/Go脚本方案是最优选择，完全满足本地部署、仅用Redis作为队列的要求，且比Kafka方案简易得多。

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