qing/blackbird
3
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

C++版本的MsgClient

Browse Source
This commit is contained in:
qing committed 2026-07-26 21:48:14 +08:00
1 parent 3d2278f6b7
commit 6efb792133
4 files changed
+545 -156
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

db/log_client.py → db/example_log_client.py
+185 -152
View File
@@ -1,152 +1,185 @@
"""
统一日志客户端 — 各服务共用。
设计要点：
- 使用 urllib.request（标准库，不引入 requests）
- 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
用法：
from log_client import logs
logs.info("dnfs", "节点新增成功 id=42")
logs.error("dnfs", "数据库连接失败", extra={"errno": 2002})
logs.push("gateway", "WARNING", "登录失败次数超限",
trace_id="req-abc", extra={"ip": "1.2.3.4"})
"""
import json
import threading
import queue
import time
import urllib.request
import urllib.error
import atexit
# ── 配置 ──────────────────────────────────────────────────────────────
DB_LOG_URL = 'http://127.0.0.1:8005/log/ingest'
_BATCH_SIZE = 50 # 队列满 50 条时 flush
_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush
_TIMEOUT = 5 # HTTP 请求超时（秒）
class LogClient:
"""异步批量日志推送客户端。"""
def __init__(self, url=DB_LOG_URL, batch_size=_BATCH_SIZE,
flush_interval=_FLUSH_INTERVAL):
self._url = url
self._batch_size = batch_size
self._flush_interval = flush_interval
self._queue = queue.Queue()
self._stop = threading.Event()
self._thread = threading.Thread(target=self._flush_loop,
daemon=True, name='log-client')
self._thread.start()
atexit.register(self._shutdown)
# ── 公开接口 ────────────────────────────────────────────────────────
def push(self, service, level, message, trace_id=None, extra=None):
"""推送一条日志（非阻塞，立即返回）。"""
self._queue.put({
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
})
def debug(self, service, message, **kw):
self.push(service, 'DEBUG', message, **kw)
def info(self, service, message, **kw):
self.push(service, 'INFO', message, **kw)
def warning(self, service, message, **kw):
self.push(service, 'WARNING', message, **kw)
def error(self, service, message, **kw):
self.push(service, 'ERROR', message, **kw)
def critical(self, service, message, **kw):
self.push(service, 'CRITICAL', message, **kw)
def flush(self):
"""手动 flush 队列中剩余的日志。"""
batch = []
while True:
try:
batch.append(self._queue.get_nowait())
if len(batch) >= self._batch_size:
self._send(batch)
batch = []
except queue.Empty:
break
if batch:
self._send(batch)
# ── 内部实现 ────────────────────────────────────────────────────────
def _flush_loop(self):
"""后台线程：积攒到 batch_size 或超时 flush_interval 时发送。"""
batch = []
last_flush = time.monotonic()
while not self._stop.is_set():
timeout = max(0, self._flush_interval - (time.monotonic() - last_flush))
try:
item = self._queue.get(timeout=timeout)
batch.append(item)
# 继续非阻塞地取，直到满 batch_size
while len(batch) < self._batch_size:
try:
batch.append(self._queue.get_nowait())
except queue.Empty:
break
except queue.Empty:
pass # 超时，没有新日志
# 满了或超时了就发
should_flush = (
len(batch) >= self._batch_size or
(batch and time.monotonic() - last_flush >= self._flush_interval)
)
if should_flush:
self._send(batch)
batch = []
last_flush = time.monotonic()
# 退出前 flush 剩余
if batch:
self._send(batch)
def _send(self, batch):
"""发送一批日志到 db 服务。失败时静默丢弃。"""
data = json.dumps({'logs': batch}).encode('utf-8')
req = urllib.request.Request(
self._url, data=data,
headers={'Content-Type': 'application/json'},
method='POST'
)
try:
with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
resp.read() # 消费响应体
except (urllib.error.URLError, urllib.error.HTTPError, OSError):
# fail-safe：日志服务不可用，静默丢弃，不影响业务
pass
except Exception:
pass
def _shutdown(self):
"""进程退出时 flush 剩余日志。"""
self._stop.set()
self.flush()
# ── 全局单例 ──────────────────────────────────────────────────────────
logs = LogClient()
"""
统一日志客户端 SDK — 各服务复制即用。
设计要点：
- 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
- 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
复制到目标服务目录后，按需修改 service_name 即可：:
from log_client import logs
logs.info("dnfs", "节点新增成功 id=42")
logs.error("dnfs", "数据库连接失败", extra={"errno": 2002})
logs.push("gateway", "WARNING", "登录失败次数超限",
trace_id="req-abc", extra={"ip": "1.2.3.4"})
"""
import json
import os as _os
import sys as _sys
import threading
import queue
import time
import atexit
from datetime import datetime, timezone
_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
from pymsg import MsgClient # noqa: E402
# ── 配置 ──────────────────────────────────────────────────────────────
_DBV2_ADDR = '127.0.0.1'
_DBV2_PORT = 8005
_BATCH_SIZE = 50 # 队列满 50 条时 flush
_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush
_TIMEOUT = 5 # socket 超时（秒）
class LogClient:
"""异步批量日志推送客户端。
通过 MsgClient（pymsg TCP 协议）逐条 INSERT 到 dbv2 的 Logs 表。
service_name 用于在 dbv2 连接中标识来源，各服务复制时自行修改。
"""
def __init__(self, service_name="example",
batch_size=_BATCH_SIZE,
flush_interval=_FLUSH_INTERVAL):
self._service_name = service_name
self._batch_size = batch_size
self._flush_interval = flush_interval
self._queue = queue.Queue()
self._stop = threading.Event()
self._thread = threading.Thread(target=self._flush_loop,
daemon=True, name='log-client')
self._thread.start()
atexit.register(self._shutdown)
# ── 公开接口 ────────────────────────────────────────────────────────
def push(self, service, level, message, trace_id=None, extra=None):
"""推送一条日志（非阻塞，立即返回）。"""
self._queue.put({
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
})
def debug(self, service, message, **kw):
self.push(service, 'DEBUG', message, **kw)
def info(self, service, message, **kw):
self.push(service, 'INFO', message, **kw)
def warning(self, service, message, **kw):
self.push(service, 'WARNING', message, **kw)
def error(self, service, message, **kw):
self.push(service, 'ERROR', message, **kw)
def critical(self, service, message, **kw):
self.push(service, 'CRITICAL', message, **kw)
def flush(self):
"""手动 flush 队列中剩余的日志。"""
batch = []
while True:
try:
batch.append(self._queue.get_nowait())
if len(batch) >= self._batch_size:
self._send(batch)
batch = []
except queue.Empty:
break
if batch:
self._send(batch)
# ── 内部实现 ────────────────────────────────────────────────────────
def _flush_loop(self):
"""后台线程：积攒到 batch_size 或超时 flush_interval 时发送。"""
batch = []
last_flush = time.monotonic()
while not self._stop.is_set():
timeout = max(0, self._flush_interval - (time.monotonic() - last_flush))
try:
item = self._queue.get(timeout=timeout)
batch.append(item)
# 继续非阻塞地取，直到满 batch_size
while len(batch) < self._batch_size:
try:
batch.append(self._queue.get_nowait())
except queue.Empty:
break
except queue.Empty:
# 超时无新日志：如果 batch 中已有积攒的条目，强制 flush
if batch:
self._send(batch)
batch = []
last_flush = time.monotonic()
continue
# 满了或超时了就发
should_flush = (
len(batch) >= self._batch_size or
(batch and time.monotonic() - last_flush >= self._flush_interval)
)
if should_flush:
self._send(batch)
batch = []
last_flush = time.monotonic()
# 退出前 flush 剩余
if batch:
self._send(batch)
def _send(self, batch):
"""通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表。"""
now = datetime.now(timezone.utc).isoformat()
try:
client = MsgClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
for entry in batch:
extra_val = entry.get("extra")
payload = {
"_route": "execute",
"action": "insert",
"table": "Logs",
"values": {
"ts": now,
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": json.dumps(extra_val, ensure_ascii=False)
if extra_val is not None else None,
},
}
resp = client.call(json.dumps(payload), timeout=_TIMEOUT)
resp_obj = json.loads(resp)
if not resp_obj.get("success"):
import sys
print(f"[log_client] INSERT 失败: {resp_obj.get('error')}",
file=sys.stderr)
except Exception as e:
import sys
print(f"[log_client] 发送失败: {e}", file=sys.stderr)
def _shutdown(self):
"""进程退出时 flush 剩余日志。"""
self._stop.set()
self.flush()
# ── 全局单例 ──────────────────────────────────────────────────────────
# 各服务复制时，把 "example" 改为自己的服务名即可。
# 例如 dnfs/log_client.py → logs = LogClient(service_name="dnfs")
# 例如 smtp/log_client.py → logs = LogClient(service_name="smtp")
logs = LogClient()
gateway/inc/DbC.hpp
+18
View File
@@ -1,10 +1,13 @@
#pragma once
#include <us/HttpClient.hpp>
#include "Msg.hpp"
#include <boost/json.hpp>
namespace qing {
namespace json = boost::json;
class DbC {
public:
// 【保留】旧版 HTTP 接口：通过 HttpClient POST 到 dbv2 HTTP /execute 端点。
// 保留用于兼容性参考，新代码优先使用 call_msg()（pymsg TCP 协议，更可靠）。
static json::value call_http(json::value& params) {
auto client = pmc::net::HttpClient();
@@ -16,6 +19,21 @@ namespace qing {
});
return json::parse(res);
}
// 新版 pymsg 接口：通过 MsgClient（TCP 长度前缀帧协议）调用 dbv2。
// 自动添加 _route = "execute"，返回 dbv2 pymsg 完整 JSON 响应
// {"success":..., "data":[...]}。
// 响应格式与 call_http 兼容，调用方可直接替换。
static json::value call_msg(const json::value& params,
double timeout = 5.0) {
json::value jv_route = params;
jv_route.as_object()["_route"] = "execute";
pmc::net::MsgClient client("127.0.0.1", 8005, "gateway");
std::string resp = client.call(
json::serialize(jv_route), timeout);
return json::parse(resp);
}
private:
static constexpr const char* _DBURL = "http://127.0.0.1:8005/execute";
};
gateway/inc/Msg.hpp
+334
View File
@@ -0,0 +1,334 @@
#pragma once
/** pymsg C++ 实现 — Length-prefixed variable-length socket message framing.
*
* Protocol: [4-byte big-endian uint32_t length][UTF-8 payload]
*
* 对应 Python 版 pymsg/pymsg/__init__.py，提供：
* - pmsg_send() / pmsg_recv() 底层协议收发函数
* - MsgClient 短连接 send / call / 长连接 listen 客户端
*
* Usage:
* #include "Msg.hpp"
*
* pmc::net::MsgClient client("127.0.0.1", 8005, "gateway");
* std::string resp = client.call(R"({"sql":"SELECT 1"})", 5.0);
* client.send("hello");
*/
#include <cerrno>
#include <cstdint>
#include <cstring>
#include <functional>
#include <stdexcept>
#include <string>
#include <vector>
#include <arpa/inet.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/select.h>
#include <sys/types.h>
#include <unistd.h>
// macOS 没有 MSG_NOSIGNAL，用 0 替代（内核不会发 SIGPIPE）
#ifndef MSG_NOSIGNAL
#define MSG_NOSIGNAL 0
#endif
namespace pmc {
namespace net {
// ============================================================================
// Exceptions
// ============================================================================
class MsgException : public std::runtime_error {
public:
explicit MsgException(const std::string &msg) : std::runtime_error(msg) {}
};
class MsgTimeoutException : public MsgException {
public:
explicit MsgTimeoutException(const std::string &msg) : MsgException(msg) {}
};
class MsgConnectionException : public MsgException {
public:
explicit MsgConnectionException(const std::string &msg) : MsgException(msg) {}
};
// ============================================================================
// Low-level protocol: 4-byte big-endian length prefix + UTF-8 payload
// ============================================================================
namespace detail {
constexpr size_t kHeaderSize = 4;
// ---- 4-byte big-endian 编码 / 解码 ----------------------------------------
inline void writeUint32BE(char *buf, uint32_t val) noexcept {
buf[0] = static_cast<char>((val >> 24) & 0xFFu);
buf[1] = static_cast<char>((val >> 16) & 0xFFu);
buf[2] = static_cast<char>((val >> 8) & 0xFFu);
buf[3] = static_cast<char>(val & 0xFFu);
}
inline uint32_t readUint32BE(const char *buf) noexcept {
return (static_cast<uint32_t>(static_cast<unsigned char>(buf[0])) << 24) |
(static_cast<uint32_t>(static_cast<unsigned char>(buf[1])) << 16) |
(static_cast<uint32_t>(static_cast<unsigned char>(buf[2])) << 8) |
(static_cast<uint32_t>(static_cast<unsigned char>(buf[3])));
}
// ---- socket 超时设置 ------------------------------------------------------
inline void setSockTimeout(int sockfd, double seconds) {
struct timeval tv {};
tv.tv_sec = static_cast<time_t>(seconds);
tv.tv_usec =
static_cast<long>((seconds - static_cast<double>(tv.tv_sec)) * 1'000'000.0);
::setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
::setsockopt(sockfd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
}
// ---- 精确读取 n 字节 ------------------------------------------------------
inline void recvExact(int sockfd, char *buf, size_t n) {
size_t received = 0;
while (received < n) {
ssize_t r = ::recv(sockfd, buf + received, n - received, 0);
if (r == 0) {
throw MsgConnectionException(
"connection closed: expected " + std::to_string(n) +
" bytes, got " + std::to_string(received));
}
if (r < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
throw MsgTimeoutException("recv timed out");
}
throw MsgConnectionException(
"recv error: " + std::string(::strerror(errno)));
}
received += static_cast<size_t>(r);
}
}
// ---- 完整发送 -------------------------------------------------------------
inline void sendAll(int sockfd, const char *data, size_t len) {
size_t sent = 0;
while (sent < len) {
ssize_t w =
::send(sockfd, data + sent, len - sent, MSG_NOSIGNAL);
if (w < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
throw MsgTimeoutException("send timed out");
}
throw MsgConnectionException(
"send error: " + std::string(::strerror(errno)));
}
sent += static_cast<size_t>(w);
}
}
// ---- 短连接 helper：connect → action → close ------------------------------
inline int openConnection(const std::string& addr, int port, double timeout) {
int fd = ::socket(AF_INET, SOCK_STREAM, 0);
if (fd < 0) {
throw MsgConnectionException(
"socket() failed: " + std::string(::strerror(errno)));
}
struct sockaddr_in serv_addr {};
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(static_cast<uint16_t>(port));
if (::inet_pton(AF_INET, addr.c_str(), &serv_addr.sin_addr) <= 0) {
::close(fd);
throw MsgConnectionException("invalid address: " + addr);
}
setSockTimeout(fd, timeout);
if (::connect(fd, reinterpret_cast<struct sockaddr *>(&serv_addr),
sizeof(serv_addr)) < 0) {
::close(fd);
throw MsgConnectionException(
"could not connect to " + addr + ":" + std::to_string(port) +
" - " + ::strerror(errno));
}
return fd;
}
} // namespace detail
// ============================================================================
// 底层协议函数：pmsg_send / pmsg_recv
// ============================================================================
/// 以 pymsg 协议发送一条字符串消息（长度前缀 + UTF-8 载荷）。
inline void pmsg_send(int sockfd, const std::string &data) {
size_t payload_size = data.size();
if (payload_size > 0xFFFFFFFFu) {
throw MsgException("payload too large: " + std::to_string(payload_size));
}
char header[detail::kHeaderSize];
detail::writeUint32BE(header, static_cast<uint32_t>(payload_size));
detail::sendAll(sockfd, header, detail::kHeaderSize);
detail::sendAll(sockfd, data.c_str(), payload_size);
}
/// 以 pymsg 协议接收一条字符串消息。
/// @throws MsgConnectionException 连接在收全消息前断开。
/// @throws MsgTimeoutException 等待超时。
inline std::string pmsg_recv(int sockfd) {
char header[detail::kHeaderSize];
detail::recvExact(sockfd, header, detail::kHeaderSize);
uint32_t length = detail::readUint32BE(header);
if (length > 64 * 1024 * 1024) { // 安全上限 64 MiB
throw MsgException("message too large: " + std::to_string(length));
}
std::string payload(static_cast<size_t>(length), '\0');
detail::recvExact(sockfd, payload.data(), length);
return payload;
}
// ============================================================================
// MsgClient
// ============================================================================
/// 基于 pymsg 协议的 TCP 客户端。
///
/// 支持三种操作模式：
/// - send() 短连接发送一条消息（不等待回复），消息带 name 前缀。
/// - call() 短连接发送 + 阻塞等待回复（RPC 模式），原始消息不添加前缀。
/// - listen() 基于 connect() 维持的长连接，持续接收消息并回调。
class MsgClient {
public:
MsgClient(const std::string &addr, int port, const std::string &name)
: _addr(addr), _port(port), _name(name), _sockfd(-1), _running(false) {}
~MsgClient() { stop(); }
// ---- 连接 ---------------------------------------------------------------
/// 创建 TCP 套接字并连接到服务器（长连接，供 listen() 使用）。
void connect() {
_sockfd = detail::openConnection(_addr, _port, /*timeout=*/10.0);
}
// ---- 发送单条消息（短连接，带名字前缀）----------------------------------
/// 短连接发送一条消息到服务器，消息格式为 ``"name: msg"``。
/// 不等待回复。短连接（每次创建新 socket，发完即关）。
void send(const std::string &msg) {
int fd = detail::openConnection(_addr, _port, /*timeout=*/3.0);
try {
pmsg_send(fd, _name + ": " + msg);
::close(fd);
} catch (...) {
::close(fd);
throw;
}
}
// ---- 请求-应答（短连接，RPC 模式）---------------------------------------
/// 短连接发送一条消息，阻塞等待并返回一条回复。
///
/// 不会给消息添加 ``name:`` 前缀（适用于 RPC/API 调用）。
///
/// @param msg 要发送的原始字符串。
/// @param timeout 超时秒数，默认 10.0。
/// @returns 服务器返回的字符串。
std::string call(const std::string &msg, double timeout = 10.0) {
int fd = detail::openConnection(_addr, _port, timeout);
try {
pmsg_send(fd, msg);
std::string resp = pmsg_recv(fd);
::close(fd);
return resp;
} catch (...) {
::close(fd);
throw;
}
}
// ---- 持续监听（长连接，阻塞回调模式）-----------------------------------
/// 在长连接上持续接收消息并通过回调处理。
///
/// 阻塞当前线程直到 stop() 被调用或连接断开。
/// 必须先调用 connect()。
///
/// @param on_message 回调 ``void(const std::string &msg)``。
void listen(std::function<void(const std::string &)> on_message) {
if (_sockfd < 0) {
throw MsgException("must call connect() before listen()");
}
_running = true;
while (_running) {
fd_set readfds;
FD_ZERO(&readfds);
FD_SET(_sockfd, &readfds);
struct timeval tv {};
tv.tv_sec = 0;
tv.tv_usec = 100'000; // 100 ms 轮询间隔
int ready =
::select(_sockfd + 1, &readfds, nullptr, nullptr, &tv);
if (ready < 0) {
if (errno == EINTR) continue;
break; // EBADF or other fatal error
}
if (ready == 0) continue; // timeout, loop around
try {
std::string text = pmsg_recv(_sockfd);
on_message(std::move(text));
} catch (const MsgConnectionException &) {
break; // 连接断开，退出循环
} catch (const MsgTimeoutException &) {
continue; // 读超时但连接仍在，继续等待
}
}
if (_sockfd >= 0) {
::close(_sockfd);
_sockfd = -1;
}
}
// ---- 停止 ---------------------------------------------------------------
/// 设置停止标志并关闭套接字。可从任意线程调用。
void stop() noexcept {
_running = false;
if (_sockfd >= 0) {
::close(_sockfd);
_sockfd = -1;
}
}
// ---- 访问器 -------------------------------------------------------------
const std::string &addr() const noexcept { return _addr; }
int port() const noexcept { return _port; }
const std::string &name() const noexcept { return _name; }
private:
std::string _addr;
int _port;
std::string _name;
int _sockfd;
bool _running;
};
} // namespace net
} // namespace pmc
gateway/src/main.cpp
+8 -4
View File
@@ -700,7 +700,9 @@ http::response<http::string_body> handleLogin(
// stored_hash = std::string(reinterpret_cast<const char*>(stmt.read_text(0)));
//}
// 新版调用db模块
// 新版调用db模块（通过 MsgClient pymsg TCP 协议）
// 【保留】旧版 HTTP 调用方式：
// auto jv_db_r = DbC::call_http(jv_select);
json::value jv_select = {
{"action", "select"},
{"table", "Users"},
@@ -708,7 +710,7 @@ http::response<http::string_body> handleLogin(
{"username", username},
}}
};
auto jv_db_r = DbC::call_http(jv_select);
auto jv_db_r = DbC::call_msg(jv_select);
auto obj_r = jv_db_r.as_object();
auto success = obj_r["success"].as_bool();
if (!success) THROW_EX_CUSTOM(HttpException, 2102, "Login (Update Step) Failed", 500, "/login");
@@ -853,7 +855,9 @@ http::response<http::string_body> handleRegister(
//stmt.bind_text(2, password_hash);
//stmt.step();
//
// 新版调用db模块
// 新版调用db模块（通过 MsgClient pymsg TCP 协议）
// 【保留】旧版 HTTP 调用方式：
// auto jv_result = DbC::call_http(jv_insert);
json::value jv_insert = {
{"action", "insert"},
{"table", "Users"},
@@ -862,7 +866,7 @@ http::response<http::string_body> handleRegister(
{"password_hash", password_hash}
}}
};
auto jv_result = DbC::call_http(jv_insert);
auto jv_result = DbC::call_msg(jv_insert);
auto obj_r = jv_result.as_object();
auto success = obj_r["success"].as_bool();
if (!success) THROW_EX_CUSTOM(HttpException, 2202, "Registe Failed", 400, "/register");