949 lines
30 KiB
C++
949 lines
30 KiB
C++
#include "RedisStreamMQ.h"
|
||
#include "../rocketmq/SimpleProducer.h"
|
||
|
||
#include <hiredis/hiredis.h>
|
||
#include <zlib.h>
|
||
|
||
#include <sys/stat.h>
|
||
#include <sys/types.h>
|
||
#include <sys/time.h>
|
||
#include <unistd.h>
|
||
#include <dirent.h>
|
||
#include <errno.h>
|
||
#include <string.h>
|
||
#include <stdarg.h>
|
||
#include <ctype.h>
|
||
|
||
#include <iostream>
|
||
#include <fstream>
|
||
#include <sstream>
|
||
#include <vector>
|
||
#include <map>
|
||
#include <set>
|
||
#include <algorithm>
|
||
#include <stdexcept>
|
||
#include <ctime>
|
||
#include <cstdlib>
|
||
#include <cstdio>
|
||
|
||
#include "../log4cplus/log4.h"
|
||
|
||
/*
|
||
* lnk20260622 修改：运行时切换版 Redis Streams 后端实现。
|
||
*
|
||
* 本文件只导出 RedisStream_xxx 函数，不再导出：
|
||
* InitializeProducer / ShutdownAndDestroyProducer
|
||
* rocketmq_producer_send
|
||
* InitializeConsumer / ShutdownAndDestroyConsumer
|
||
* rocketmq_consumer_receive
|
||
*
|
||
* 这些统一入口应在 SimpleProducer.cpp 中按配置文件 [MQ] Backend 转发，
|
||
* 否则同时编译 RocketMQ 和 RedisStream 时会发生重复定义链接错误。
|
||
*/
|
||
|
||
extern std::string G_ROCKETMQ_PRODUCER; // 保留原配置名；Redis 模式下仅用于日志
|
||
extern std::string G_ROCKETMQ_IPPORT; // Redis 模式下建议填 127.0.0.1:6379[/db]
|
||
extern std::string G_MQCONSUMER_IPPORT; // Redis 模式下消费者连接地址
|
||
extern std::string G_ROCKETMQ_CONSUMER; // Redis consumer group
|
||
extern std::string FRONT_INST;
|
||
|
||
extern int g_front_seg_index;
|
||
extern char subdir[128];
|
||
|
||
static const int64_t REDIS_BLOCK_MS = 1000; //XREADGROUP 阻塞等待时间
|
||
static const int64_t REDIS_IDLE_CLAIM_MS = 60000; //消息空闲多久被其他消费者抢占
|
||
static const int REDIS_READ_COUNT = 16; //每次 XREADGROUP 的最大读取数量
|
||
static const size_t GZIP_THRESHOLD_BYTES = 1024; //超过这个大小的消息才压缩
|
||
static const char* SPOOL_DIR = "/FeProject/data/redisstream_spool"; //本地缓存目录
|
||
|
||
static int64_t now_ms() //当前时间，单位毫秒
|
||
{
|
||
struct timeval tv;
|
||
gettimeofday(&tv, NULL);
|
||
return (int64_t)tv.tv_sec * 1000 + tv.tv_usec / 1000;
|
||
}
|
||
|
||
/*
|
||
* 注意：如果 RocketMQ 原 SimpleProducer.cpp 已经定义了 G_APP_START_MS / G_START_SKEW_MS，
|
||
* 这里不要再定义全局同名变量，避免多后端同时编译时重复定义。
|
||
*/
|
||
static int64_t redis_app_start_ms()//RedisStream 后端的程序启动时间
|
||
{
|
||
static int64_t v = now_ms();
|
||
return v;
|
||
}
|
||
|
||
static const int64_t REDIS_START_SKEW_MS = 1000;
|
||
|
||
static std::string to_string_i64(int64_t v)
|
||
{
|
||
char buf[64];
|
||
snprintf(buf, sizeof(buf), "%lld", (long long)v);
|
||
return std::string(buf);
|
||
}
|
||
|
||
static std::string shell_safe_name(const std::string& s)
|
||
{
|
||
std::string r;
|
||
for (size_t i = 0; i < s.size(); ++i) {
|
||
unsigned char c = (unsigned char)s[i];
|
||
if ((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') ||
|
||
(c >= '0' && c <= '9') || c == '_' || c == '-' || c == '.') {
|
||
r.push_back((char)c);
|
||
} else {
|
||
r.push_back('_');
|
||
}
|
||
}
|
||
return r.empty() ? "empty" : r;
|
||
}
|
||
|
||
static std::string build_stream_name(const std::string& topic, const std::string& env)//构造stream名称，格式：env_shortTopic
|
||
{
|
||
if (!topic.empty() && topic.find('%') != std::string::npos) {
|
||
std::string shortTopic = topic.substr(topic.find('%') + 1);
|
||
if (!env.empty()) return env + "_" + shortTopic;
|
||
return shortTopic;
|
||
}
|
||
|
||
if (!env.empty()) {
|
||
const std::string prefix = env + "_";
|
||
if (topic.find(prefix) == 0) return topic;
|
||
return prefix + topic;
|
||
}
|
||
return topic;
|
||
}
|
||
|
||
static void ensure_dir(const std::string& dir)//创建目录，如果不存在
|
||
{
|
||
if (dir.empty()) return;
|
||
if (access(dir.c_str(), F_OK) == 0) return;
|
||
if (mkdir(dir.c_str(), 0755) != 0 && errno != EEXIST) {
|
||
std::cerr << "[REDIS_STREAM][SPOOL] mkdir failed: " << dir
|
||
<< " errno=" << errno << " " << strerror(errno) << std::endl;
|
||
}
|
||
}
|
||
|
||
static std::string base64_encode(const std::string& input)//编码为 base64
|
||
{
|
||
static const char* table =
|
||
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
|
||
std::string out;
|
||
int val = 0;
|
||
int valb = -6;
|
||
for (size_t i = 0; i < input.size(); ++i) {
|
||
unsigned char c = (unsigned char)input[i];
|
||
val = (val << 8) + c;
|
||
valb += 8;
|
||
while (valb >= 0) {
|
||
out.push_back(table[(val >> valb) & 0x3F]);
|
||
valb -= 6;
|
||
}
|
||
}
|
||
if (valb > -6) out.push_back(table[((val << 8) >> (valb + 8)) & 0x3F]);
|
||
while (out.size() % 4) out.push_back('=');
|
||
return out;
|
||
}
|
||
|
||
static std::string gzip_compress(const std::string& input)//压缩字符串，返回 gzip 格式
|
||
{
|
||
if (input.empty()) return std::string();
|
||
|
||
z_stream zs;
|
||
memset(&zs, 0, sizeof(zs));
|
||
|
||
int ret = deflateInit2(&zs, Z_BEST_SPEED, Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY);
|
||
if (ret != Z_OK) {
|
||
throw std::runtime_error("deflateInit2 failed");
|
||
}
|
||
|
||
zs.next_in = (Bytef*)input.data();
|
||
zs.avail_in = (uInt)input.size();
|
||
|
||
char outbuffer[32768];
|
||
std::string out;
|
||
|
||
do {
|
||
zs.next_out = reinterpret_cast<Bytef*>(outbuffer);
|
||
zs.avail_out = sizeof(outbuffer);
|
||
|
||
ret = deflate(&zs, Z_FINISH);
|
||
|
||
if (out.size() < zs.total_out) {
|
||
out.append(outbuffer, zs.total_out - out.size());
|
||
}
|
||
} while (ret == Z_OK);
|
||
|
||
deflateEnd(&zs);
|
||
|
||
if (ret != Z_STREAM_END) {
|
||
throw std::runtime_error("gzip deflate failed");
|
||
}
|
||
|
||
return out;
|
||
}
|
||
|
||
static bool gzip_decompress(const std::string& input, std::string& output) //解压缩 gzip 格式字符串
|
||
{
|
||
output.clear();
|
||
if (input.empty()) return true;
|
||
|
||
z_stream zs;
|
||
memset(&zs, 0, sizeof(zs));
|
||
|
||
int ret = inflateInit2(&zs, 15 + 32);
|
||
if (ret != Z_OK) return false;
|
||
|
||
zs.next_in = (Bytef*)input.data();
|
||
zs.avail_in = (uInt)input.size();
|
||
|
||
char outbuffer[32768];
|
||
|
||
do {
|
||
zs.next_out = reinterpret_cast<Bytef*>(outbuffer);
|
||
zs.avail_out = sizeof(outbuffer);
|
||
|
||
ret = inflate(&zs, 0);
|
||
|
||
if (output.size() < zs.total_out) {
|
||
output.append(outbuffer, zs.total_out - output.size());
|
||
}
|
||
|
||
if (ret == Z_STREAM_ERROR || ret == Z_DATA_ERROR || ret == Z_MEM_ERROR) {
|
||
inflateEnd(&zs);
|
||
return false;
|
||
}
|
||
} while (ret != Z_STREAM_END);
|
||
|
||
inflateEnd(&zs);
|
||
return true;
|
||
}
|
||
|
||
static int base64_value(char c)
|
||
{
|
||
if (c >= 'A' && c <= 'Z') return c - 'A';
|
||
if (c >= 'a' && c <= 'z') return c - 'a' + 26;
|
||
if (c >= '0' && c <= '9') return c - '0' + 52;
|
||
if (c == '+') return 62;
|
||
if (c == '/') return 63;
|
||
return -1;
|
||
}
|
||
|
||
static bool base64_decode(const std::string& input, std::string& output) //解码 base64
|
||
{
|
||
output.clear();
|
||
int val = 0;
|
||
int valb = -8;
|
||
for (size_t i = 0; i < input.size(); ++i) {
|
||
char c = input[i];
|
||
if (c == '=') break;
|
||
int d = base64_value(c);
|
||
if (d == -1) {
|
||
if (c == '\r' || c == '\n' || c == ' ' || c == '\t') continue;
|
||
return false;
|
||
}
|
||
val = (val << 6) + d;
|
||
valb += 6;
|
||
if (valb >= 0) {
|
||
output.push_back(char((val >> valb) & 0xFF));
|
||
valb -= 8;
|
||
}
|
||
}
|
||
return true;
|
||
}
|
||
|
||
static bool decode_body(const std::string& enc, //解压缩
|
||
const std::string& body,
|
||
std::string& out)
|
||
{
|
||
if (enc == "gzip") {
|
||
std::string gz;
|
||
if (!base64_decode(body, gz)) return false;
|
||
return gzip_decompress(gz, out);
|
||
}
|
||
|
||
out = body;
|
||
return true;
|
||
}
|
||
|
||
static void encode_body(const std::string& json,
|
||
std::string& enc,
|
||
std::string& body)
|
||
{
|
||
if (json.size() >= GZIP_THRESHOLD_BYTES) {
|
||
enc = "gzip";
|
||
body = base64_encode(gzip_compress(json)); //大的压缩后编码
|
||
} else {
|
||
enc = "plain";
|
||
body = json; //小的保持json
|
||
}
|
||
}
|
||
|
||
static std::string parse_host(const std::string& addr)//解析主机地址，去掉协议、端口和数据库部分
|
||
{
|
||
std::string s = addr;
|
||
if (s.find("redis://") == 0) s = s.substr(8);
|
||
size_t slash = s.find('/');
|
||
if (slash != std::string::npos) s = s.substr(0, slash);
|
||
size_t colon = s.rfind(':');
|
||
if (colon == std::string::npos) return s;
|
||
return s.substr(0, colon);
|
||
}
|
||
|
||
static int parse_port(const std::string& addr)//提取端口
|
||
{
|
||
std::string s = addr;
|
||
if (s.find("redis://") == 0) s = s.substr(8);
|
||
size_t slash = s.find('/');
|
||
if (slash != std::string::npos) s = s.substr(0, slash);
|
||
size_t colon = s.rfind(':');
|
||
if (colon == std::string::npos) return 6379;
|
||
return atoi(s.substr(colon + 1).c_str());
|
||
}
|
||
|
||
static int parse_db(const std::string& addr) //提取数据库
|
||
{
|
||
size_t slash = addr.rfind('/');
|
||
if (slash == std::string::npos) return -1;
|
||
std::string db = addr.substr(slash + 1);
|
||
if (db.empty()) return -1;
|
||
for (size_t i = 0; i < db.size(); ++i) {
|
||
if (!isdigit((unsigned char)db[i])) return -1;
|
||
}
|
||
return atoi(db.c_str());
|
||
}
|
||
|
||
class RedisConnection { //Redis 连接封装
|
||
public:
|
||
RedisConnection() : ctx_(NULL), db_(-1) {}
|
||
~RedisConnection() { close(); }
|
||
|
||
void configure(const std::string& addr)
|
||
{
|
||
addr_ = addr;
|
||
host_ = parse_host(addr);
|
||
port_ = parse_port(addr);
|
||
db_ = parse_db(addr);
|
||
if (host_.empty()) host_ = "127.0.0.1";
|
||
}
|
||
|
||
redisContext* get()
|
||
{
|
||
if (ctx_ && ctx_->err == 0) return ctx_;
|
||
reconnect();
|
||
return ctx_;
|
||
}
|
||
|
||
redisReply* command(const char* fmt, ...)
|
||
{
|
||
redisContext* c = get();
|
||
if (!c) return NULL;
|
||
|
||
va_list ap;
|
||
va_start(ap, fmt);
|
||
void* r = redisvCommand(c, fmt, ap);
|
||
va_end(ap);
|
||
|
||
if (!r) {
|
||
close();
|
||
return NULL;
|
||
}
|
||
return (redisReply*)r;
|
||
}
|
||
|
||
bool ping()
|
||
{
|
||
redisReply* r = command("PING");
|
||
if (!r) return false;
|
||
bool ok = (r->type == REDIS_REPLY_STATUS && r->str && std::string(r->str) == "PONG");
|
||
freeReplyObject(r);
|
||
return ok;
|
||
}
|
||
|
||
void close()
|
||
{
|
||
if (ctx_) {
|
||
redisFree(ctx_);
|
||
ctx_ = NULL;
|
||
}
|
||
}
|
||
|
||
private:
|
||
void reconnect()
|
||
{
|
||
close();
|
||
|
||
struct timeval tv;
|
||
tv.tv_sec = 1;
|
||
tv.tv_usec = 500000;
|
||
|
||
ctx_ = redisConnectWithTimeout(host_.c_str(), port_, tv);
|
||
if (!ctx_ || ctx_->err) {
|
||
if (ctx_) {
|
||
std::cerr << "[REDIS_STREAM] connect failed: " << ctx_->errstr << std::endl;
|
||
close();
|
||
} else {
|
||
std::cerr << "[REDIS_STREAM] connect failed: null context" << std::endl;
|
||
}
|
||
return;
|
||
}
|
||
|
||
redisSetTimeout(ctx_, tv);
|
||
|
||
if (db_ >= 0) {
|
||
redisReply* r = (redisReply*)redisCommand(ctx_, "SELECT %d", db_);
|
||
if (!r || (r->type == REDIS_REPLY_ERROR)) {
|
||
std::cerr << "[REDIS_STREAM] select db failed: " << db_ << std::endl;
|
||
if (r) freeReplyObject(r);
|
||
close();
|
||
return;
|
||
}
|
||
freeReplyObject(r);
|
||
}
|
||
}
|
||
|
||
private:
|
||
redisContext* ctx_;
|
||
std::string addr_;
|
||
std::string host_;
|
||
int port_;
|
||
int db_;
|
||
};
|
||
|
||
class RedisStreamProducer { //RedisStream 生产者封装
|
||
public:
|
||
explicit RedisStreamProducer(const std::string& addr)
|
||
{
|
||
conn_.configure(addr);
|
||
ensure_dir(SPOOL_DIR);
|
||
if (!conn_.ping()) {
|
||
std::cerr << "[REDIS_STREAM][PRODUCER] Redis not ready, spool enabled" << std::endl;
|
||
}
|
||
}
|
||
|
||
void sendMessage(const std::string& json,
|
||
const std::string& topic,
|
||
const std::string& tag,
|
||
const std::string& key)
|
||
{
|
||
flushSpool();
|
||
|
||
std::string enc;
|
||
std::string body; //序列化的数据json
|
||
encode_body(json, enc, body); //压缩
|
||
|
||
const std::string stream = build_stream_name(topic, tag); //构造stream名称，格式：env_shortTopic
|
||
const std::string ts = to_string_i64(now_ms());
|
||
|
||
if (!xadd(stream, key, tag, enc, body, ts)) { //
|
||
spool(stream, key, tag, enc, body, ts);
|
||
throw std::runtime_error("Redis XADD failed, message spooled");
|
||
}
|
||
}
|
||
|
||
void shutdown() {}
|
||
|
||
private:
|
||
bool xadd(const std::string& stream,
|
||
const std::string& key,
|
||
const std::string& tag,
|
||
const std::string& enc,
|
||
const std::string& body,
|
||
const std::string& ts)
|
||
{
|
||
redisReply* r = conn_.command(
|
||
"XADD %b * key %b tag %b enc %b body %b ts %b",
|
||
stream.data(), stream.size(),
|
||
key.data(), key.size(),
|
||
tag.data(), tag.size(),
|
||
enc.data(), enc.size(),
|
||
body.data(), body.size(),
|
||
ts.data(), ts.size());
|
||
|
||
if (!r) return false;
|
||
|
||
bool ok = (r->type == REDIS_REPLY_STRING);
|
||
if (!ok && r->type == REDIS_REPLY_ERROR) {
|
||
std::cerr << "[REDIS_STREAM][XADD_ERROR] " << (r->str ? r->str : "") << std::endl;
|
||
}
|
||
freeReplyObject(r);
|
||
return ok;
|
||
}
|
||
|
||
void spool(const std::string& stream,
|
||
const std::string& key,
|
||
const std::string& tag,
|
||
const std::string& enc,
|
||
const std::string& body,
|
||
const std::string& ts)
|
||
{
|
||
ensure_dir(SPOOL_DIR);
|
||
std::string fn = std::string(SPOOL_DIR) + "/" + to_string_i64(now_ms()) + "_" +
|
||
shell_safe_name(stream) + "_" + shell_safe_name(key) + ".msg.tmp";
|
||
std::string done = fn.substr(0, fn.size() - 4);
|
||
|
||
std::ofstream out(fn.c_str(), std::ios::out | std::ios::binary);
|
||
if (!out) {
|
||
std::cerr << "[REDIS_STREAM][SPOOL] open failed: " << fn << std::endl;
|
||
return;
|
||
}
|
||
|
||
out << "stream=" << stream << "\n";
|
||
out << "key=" << key << "\n";
|
||
out << "tag=" << tag << "\n";
|
||
out << "enc=" << enc << "\n";
|
||
out << "ts=" << ts << "\n";
|
||
out << "body=" << body << "\n";
|
||
out.close();
|
||
|
||
if (rename(fn.c_str(), done.c_str()) != 0) {
|
||
std::cerr << "[REDIS_STREAM][SPOOL] rename failed errno=" << errno << std::endl;
|
||
}
|
||
}
|
||
|
||
static bool parse_spool_file(const std::string& path,
|
||
std::string& stream,
|
||
std::string& key,
|
||
std::string& tag,
|
||
std::string& enc,
|
||
std::string& body,
|
||
std::string& ts)
|
||
{
|
||
std::ifstream in(path.c_str(), std::ios::in | std::ios::binary);
|
||
if (!in) return false;
|
||
|
||
std::string line;
|
||
while (std::getline(in, line)) {
|
||
if (line.find("stream=") == 0) stream = line.substr(7);
|
||
else if (line.find("key=") == 0) key = line.substr(4);
|
||
else if (line.find("tag=") == 0) tag = line.substr(4);
|
||
else if (line.find("enc=") == 0) enc = line.substr(4);
|
||
else if (line.find("ts=") == 0) ts = line.substr(3);
|
||
else if (line.find("body=") == 0) {
|
||
body = line.substr(5);
|
||
std::string rest;
|
||
while (std::getline(in, rest)) {
|
||
body += "\n";
|
||
body += rest;
|
||
}
|
||
break;
|
||
}
|
||
}
|
||
return !stream.empty() && !enc.empty() && !ts.empty();
|
||
}
|
||
|
||
void flushSpool()
|
||
{
|
||
DIR* dir = opendir(SPOOL_DIR);
|
||
if (!dir) return;
|
||
|
||
std::vector<std::string> files;
|
||
struct dirent* ent = NULL;
|
||
while ((ent = readdir(dir)) != NULL) {
|
||
std::string name = ent->d_name;
|
||
if (name == "." || name == "..") continue;
|
||
if (name.size() >= 4 && name.substr(name.size() - 4) == ".tmp") continue;
|
||
files.push_back(std::string(SPOOL_DIR) + "/" + name);
|
||
}
|
||
closedir(dir);
|
||
|
||
std::sort(files.begin(), files.end());
|
||
|
||
for (size_t i = 0; i < files.size(); ++i) {
|
||
std::string stream, key, tag, enc, body, ts;
|
||
if (!parse_spool_file(files[i], stream, key, tag, enc, body, ts)) {
|
||
continue;
|
||
}
|
||
if (xadd(stream, key, tag, enc, body, ts)) {
|
||
unlink(files[i].c_str());
|
||
std::cout << "[REDIS_STREAM][SPOOL_FLUSHED] " << files[i] << std::endl;
|
||
} else {
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
private:
|
||
RedisConnection conn_;
|
||
};
|
||
|
||
class RedisStreamConsumer { //RedisStream 消费者封装
|
||
public:
|
||
RedisStreamConsumer(const std::string& group, const std::string& addr)
|
||
: group_(group), stop_(false)
|
||
{
|
||
conn_.configure(addr);
|
||
if (group_.empty()) group_ = "front_default";
|
||
|
||
char buf[64];
|
||
snprintf(buf, sizeof(buf), "_%s_%d", subdir, g_front_seg_index);
|
||
|
||
group_ += buf;
|
||
|
||
consumer_ = group_ + "_" + to_string_i64((int64_t)getpid()) + "_" + to_string_i64(now_ms());
|
||
}
|
||
|
||
void subscribe(const std::string& topic,
|
||
const std::string& tag,
|
||
MessageCallBack cb)
|
||
{
|
||
Sub s;
|
||
s.topic = topic;
|
||
s.tag = tag;
|
||
s.stream = build_stream_name(topic, tag);
|
||
s.cb = cb;
|
||
subs_.push_back(s);
|
||
|
||
callbacks_[std::make_pair(topic, tag)] = cb;
|
||
callbacks_[std::make_pair(s.stream, tag)] = cb;
|
||
|
||
size_t pos = topic.find('%');
|
||
if (pos != std::string::npos) {
|
||
std::string shortTopic = topic.substr(pos + 1);
|
||
callbacks_[std::make_pair(shortTopic, tag)] = cb;
|
||
}
|
||
|
||
createGroup(s.stream);
|
||
|
||
std::cout << "[REDIS_STREAM][SUB] stream=" << s.stream
|
||
<< " topic=" << topic << " tag=" << tag
|
||
<< " group=" << group_ << " consumer=" << consumer_ << std::endl;
|
||
}
|
||
|
||
void start()
|
||
{
|
||
stop_ = false;
|
||
while (!stop_) {
|
||
claimOnce();
|
||
readOnce();
|
||
}
|
||
}
|
||
|
||
void shutdown()
|
||
{
|
||
stop_ = true;
|
||
conn_.close();
|
||
}
|
||
|
||
private:
|
||
struct Sub {
|
||
std::string topic;
|
||
std::string tag;
|
||
std::string stream;
|
||
MessageCallBack cb;
|
||
};
|
||
|
||
void createGroup(const std::string& stream)
|
||
{
|
||
redisReply* r = conn_.command("XGROUP CREATE %b %b $ MKSTREAM",
|
||
stream.data(), stream.size(),
|
||
group_.data(), group_.size());
|
||
if (!r) {
|
||
std::cerr << "[REDIS_STREAM][GROUP] create failed: " << stream << std::endl;
|
||
return;
|
||
}
|
||
|
||
if (r->type == REDIS_REPLY_ERROR) {
|
||
std::string err = r->str ? r->str : "";
|
||
if (err.find("BUSYGROUP") == std::string::npos) {
|
||
std::cerr << "[REDIS_STREAM][GROUP_ERROR] " << stream << " " << err << std::endl;
|
||
}
|
||
}
|
||
freeReplyObject(r);
|
||
}
|
||
|
||
bool parseFields(redisReply* arr,
|
||
std::map<std::string, std::string>& fields)
|
||
{
|
||
fields.clear();
|
||
if (!arr || arr->type != REDIS_REPLY_ARRAY) return false;
|
||
for (size_t i = 0; i + 1 < arr->elements; i += 2) {
|
||
redisReply* k = arr->element[i];
|
||
redisReply* v = arr->element[i + 1];
|
||
if (!k || !v || k->type != REDIS_REPLY_STRING || v->type != REDIS_REPLY_STRING) continue;
|
||
fields[std::string(k->str, k->len)] = std::string(v->str, v->len);
|
||
}
|
||
return true;
|
||
}
|
||
|
||
static std::string getField(const std::map<std::string, std::string>& m,
|
||
const std::string& k)
|
||
{
|
||
std::map<std::string, std::string>::const_iterator it = m.find(k);
|
||
if (it == m.end()) return "";
|
||
return it->second;
|
||
}
|
||
|
||
rocketmq::MQMessageExt makeMsg(const Sub& sub,
|
||
const std::string& id,
|
||
const std::map<std::string, std::string>& fields)
|
||
{
|
||
std::string enc = getField(fields, "enc");
|
||
std::string bodyRaw = getField(fields, "body");
|
||
std::string body;
|
||
if (!decode_body(enc, bodyRaw, body)) {
|
||
throw std::runtime_error("decode body failed");
|
||
}
|
||
|
||
std::string key = getField(fields, "key");
|
||
std::string tag = getField(fields, "tag");
|
||
std::string ts = getField(fields, "ts");
|
||
|
||
if (tag.empty()) tag = sub.tag;
|
||
|
||
int64_t born = ts.empty() ? now_ms() : atoll(ts.c_str());
|
||
|
||
return rocketmq::MQMessageExt(sub.topic, tag, key, body, born, id);
|
||
}
|
||
|
||
MessageCallBack findCallback(const std::string& topic,
|
||
const std::string& stream,
|
||
const std::string& tag)
|
||
{
|
||
std::map<std::pair<std::string, std::string>, MessageCallBack>::iterator it;
|
||
it = callbacks_.find(std::make_pair(topic, tag));
|
||
if (it != callbacks_.end()) return it->second;
|
||
|
||
it = callbacks_.find(std::make_pair(stream, tag));
|
||
if (it != callbacks_.end()) return it->second;
|
||
|
||
return NULL;
|
||
}
|
||
|
||
void ack(const std::string& stream, const std::string& id)
|
||
{
|
||
redisReply* r = conn_.command("XACK %b %b %b",
|
||
stream.data(), stream.size(),
|
||
group_.data(), group_.size(),
|
||
id.data(), id.size());
|
||
if (r) freeReplyObject(r);
|
||
}
|
||
|
||
void handleOne(const Sub& sub, const std::string& id, redisReply* fieldsReply)
|
||
{
|
||
try {
|
||
std::map<std::string, std::string> fields;
|
||
if (!parseFields(fieldsReply, fields)) {
|
||
std::cerr << "[REDIS_STREAM][BAD_MSG] no fields stream=" << sub.stream << " id=" << id << std::endl;
|
||
ack(sub.stream, id);
|
||
return;
|
||
}
|
||
|
||
rocketmq::MQMessageExt msg = makeMsg(sub, id, fields);
|
||
|
||
MessageCallBack cb = findCallback(sub.topic, sub.stream, msg.getTags());
|
||
if (!cb) {
|
||
std::cout << "[REDIS_STREAM][NO_CALLBACK] stream=" << sub.stream
|
||
<< " tag=" << msg.getTags() << std::endl;
|
||
ack(sub.stream, id);
|
||
return;
|
||
}
|
||
|
||
rocketmq::ConsumeStatus ret = cb(msg);
|
||
if (ret == rocketmq::CONSUME_SUCCESS) {
|
||
ack(sub.stream, id);
|
||
} else {
|
||
std::cout << "[REDIS_STREAM][RETRY_LATER] stream=" << sub.stream
|
||
<< " id=" << id << std::endl;
|
||
}
|
||
} catch (const std::exception& e) {
|
||
std::cerr << "[REDIS_STREAM][HANDLE_EXCEPTION] " << e.what()
|
||
<< " stream=" << sub.stream << " id=" << id << std::endl;
|
||
} catch (...) {
|
||
std::cerr << "[REDIS_STREAM][HANDLE_UNKNOWN_EXCEPTION] stream="
|
||
<< sub.stream << " id=" << id << std::endl;
|
||
}
|
||
}
|
||
|
||
void readOnce()
|
||
{
|
||
for (size_t i = 0; i < subs_.size(); ++i) {
|
||
Sub& sub = subs_[i];
|
||
|
||
redisReply* r = conn_.command(
|
||
"XREADGROUP GROUP %b %b COUNT %d BLOCK %lld STREAMS %b >",
|
||
group_.data(), group_.size(),
|
||
consumer_.data(), consumer_.size(),
|
||
REDIS_READ_COUNT,
|
||
(long long)REDIS_BLOCK_MS,
|
||
sub.stream.data(), sub.stream.size());
|
||
|
||
if (!r) {
|
||
usleep(500000);
|
||
continue;
|
||
}
|
||
|
||
if (r->type == REDIS_REPLY_NIL) {
|
||
freeReplyObject(r);
|
||
continue;
|
||
}
|
||
|
||
if (r->type == REDIS_REPLY_ERROR) {
|
||
std::string err = r->str ? r->str : "";
|
||
std::cerr << "[REDIS_STREAM][XREADGROUP_ERROR] " << err << std::endl;
|
||
if (err.find("NOGROUP") != std::string::npos) {
|
||
createGroup(sub.stream);
|
||
}
|
||
freeReplyObject(r);
|
||
continue;
|
||
}
|
||
|
||
if (r->type == REDIS_REPLY_ARRAY) {
|
||
for (size_t si = 0; si < r->elements; ++si) {
|
||
redisReply* streamPair = r->element[si];
|
||
if (!streamPair || streamPair->type != REDIS_REPLY_ARRAY || streamPair->elements < 2) continue;
|
||
|
||
redisReply* messages = streamPair->element[1];
|
||
if (!messages || messages->type != REDIS_REPLY_ARRAY) continue;
|
||
|
||
for (size_t mi = 0; mi < messages->elements; ++mi) {
|
||
redisReply* msgPair = messages->element[mi];
|
||
if (!msgPair || msgPair->type != REDIS_REPLY_ARRAY || msgPair->elements < 2) continue;
|
||
redisReply* idReply = msgPair->element[0];
|
||
redisReply* fieldsReply = msgPair->element[1];
|
||
if (!idReply || idReply->type != REDIS_REPLY_STRING) continue;
|
||
|
||
std::string id(idReply->str, idReply->len);
|
||
handleOne(sub, id, fieldsReply);
|
||
}
|
||
}
|
||
}
|
||
|
||
freeReplyObject(r);
|
||
}
|
||
}
|
||
|
||
void claimOnce()
|
||
{
|
||
for (size_t i = 0; i < subs_.size(); ++i) {
|
||
Sub& sub = subs_[i];
|
||
|
||
redisReply* r = conn_.command(
|
||
"XAUTOCLAIM %b %b %b %lld 0-0 COUNT %d",
|
||
sub.stream.data(), sub.stream.size(),
|
||
group_.data(), group_.size(),
|
||
consumer_.data(), consumer_.size(),
|
||
(long long)REDIS_IDLE_CLAIM_MS,
|
||
REDIS_READ_COUNT);
|
||
|
||
if (!r) continue;
|
||
|
||
if (r->type == REDIS_REPLY_ERROR) {
|
||
std::string err = r->str ? r->str : "";
|
||
if (err.find("NOGROUP") != std::string::npos) createGroup(sub.stream);
|
||
freeReplyObject(r);
|
||
continue;
|
||
}
|
||
|
||
if (r->type == REDIS_REPLY_ARRAY && r->elements >= 2) {
|
||
redisReply* messages = r->element[1];
|
||
if (messages && messages->type == REDIS_REPLY_ARRAY) {
|
||
for (size_t mi = 0; mi < messages->elements; ++mi) {
|
||
redisReply* msgPair = messages->element[mi];
|
||
if (!msgPair || msgPair->type != REDIS_REPLY_ARRAY || msgPair->elements < 2) continue;
|
||
redisReply* idReply = msgPair->element[0];
|
||
redisReply* fieldsReply = msgPair->element[1];
|
||
if (!idReply || idReply->type != REDIS_REPLY_STRING) continue;
|
||
|
||
std::string id(idReply->str, idReply->len);
|
||
std::cout << "[REDIS_STREAM][CLAIM] stream=" << sub.stream
|
||
<< " id=" << id << std::endl;
|
||
handleOne(sub, id, fieldsReply);
|
||
}
|
||
}
|
||
}
|
||
|
||
freeReplyObject(r);
|
||
}
|
||
}
|
||
|
||
private:
|
||
RedisConnection conn_;
|
||
std::string group_;
|
||
std::string consumer_;
|
||
bool stop_;
|
||
std::vector<Sub> subs_;
|
||
std::map<std::pair<std::string, std::string>, MessageCallBack> callbacks_;
|
||
};
|
||
|
||
static RedisStreamProducer* g_redis_producer = NULL; //全局生产者实例
|
||
static RedisStreamConsumer* g_redis_consumer = NULL; //全局消费者实例
|
||
|
||
bool RedisStream_should_process_after_start(const rocketmq::MQMessageExt& msg) //判断消息是否在程序启动后产生
|
||
{
|
||
const int64_t born_ts = static_cast<int64_t>(msg.getBornTimestamp());
|
||
return born_ts >= (redis_app_start_ms() - REDIS_START_SKEW_MS);
|
||
}
|
||
|
||
void RedisStream_InitializeProducer() //初始化生产者
|
||
{
|
||
if (g_redis_producer == NULL) {
|
||
std::string addr = G_ROCKETMQ_IPPORT.empty() ? G_MQCONSUMER_IPPORT : G_ROCKETMQ_IPPORT;
|
||
g_redis_producer = new RedisStreamProducer(addr);
|
||
std::cout << "[REDIS_STREAM] producer initialized addr=" << addr << std::endl;
|
||
}
|
||
}
|
||
|
||
void RedisStream_ShutdownAndDestroyProducer()//关闭生产者
|
||
{
|
||
if (g_redis_producer != NULL) {
|
||
g_redis_producer->shutdown();
|
||
delete g_redis_producer;
|
||
g_redis_producer = NULL;
|
||
}
|
||
}
|
||
|
||
void RedisStream_producer_send(const std::string& body, //发送消息
|
||
const std::string& topic,
|
||
const std::string& tags,
|
||
const std::string& keys)
|
||
{
|
||
if (g_redis_producer == NULL) RedisStream_InitializeProducer();
|
||
|
||
try {
|
||
g_redis_producer->sendMessage(body, topic, tags, keys);
|
||
} catch (const std::exception& e) {
|
||
std::cerr << "[REDIS_STREAM][SEND_FAIL] " << e.what() << std::endl;
|
||
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ,
|
||
"【ERROR】前置的%s%d号进程 redis stream发送失败,已尝试本地落盘",
|
||
get_front_msg_from_subdir(), g_front_seg_index);
|
||
} catch (...) {
|
||
std::cerr << "[REDIS_STREAM][SEND_FAIL] unknown exception" << std::endl;
|
||
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ,
|
||
"【ERROR】前置的%s%d号进程 redis stream发送未知异常",
|
||
get_front_msg_from_subdir(), g_front_seg_index);
|
||
}
|
||
}
|
||
|
||
void RedisStream_InitializeConsumer(const std::string& consumerName, //初始化消费者
|
||
const std::string& nameServer,
|
||
const std::vector<Subscription>& subscriptions)
|
||
{
|
||
if (g_redis_consumer == NULL) {
|
||
g_redis_consumer = new RedisStreamConsumer(consumerName, nameServer);
|
||
for (size_t i = 0; i < subscriptions.size(); ++i) {
|
||
g_redis_consumer->subscribe(subscriptions[i].topic,
|
||
subscriptions[i].tag,
|
||
subscriptions[i].callback);
|
||
}
|
||
}
|
||
}
|
||
|
||
void RedisStream_ShutdownAndDestroyConsumer()//关闭消费者
|
||
{
|
||
if (g_redis_consumer != NULL) {
|
||
g_redis_consumer->shutdown();
|
||
delete g_redis_consumer;
|
||
g_redis_consumer = NULL;
|
||
}
|
||
}
|
||
|
||
void RedisStream_consumer_receive(const std::string& consumerName, //启动消费者接收消息
|
||
const std::string& nameServer,
|
||
const std::vector<Subscription>& subscriptions)
|
||
{
|
||
RedisStream_InitializeConsumer(consumerName, nameServer, subscriptions);
|
||
if (g_redis_consumer) {
|
||
g_redis_consumer->start();
|
||
}
|
||
}
|