[微服务即时通讯系统]消息存储子服务的实现与测试
·
本专栏内容为：项目专栏
💓博主csdn个人主页：小小unicorn
⏩专栏分类：微服务即时通讯系统
🚚代码仓库：小小unicorn的代码仓库🚚
🌹🌹🌹关注我带你学习编程知识
消息存储子服务
功能设计
消息管理子服务，主要用于管理消息的存储：
• 文本消息，储存在 ElasticSearch 文档搜索服务中。
• 文件/语音/图片，需要转储到文件管理子服务中。
除了管理消息的存储，还需要管理消息的搜索获取，因此需要对外提供以下接口：
- 获取历史消息：
a. 获取最近N 条消息：用于登录成功后，点击对方头像打开聊天框时显示最近
的消息
b. 获取指定时间段内的消息：用户可以进行聊天消息的按时间搜索 - 关键字消息搜索：用户可以针对指定好友的聊天进行聊天消息的关键字搜索
模块划分
- 参数/配置文件解析模块：基于
gflags框架直接使用进行参数/配置文件解析。 - 日志模块：基于
spdlog框架封装的模块直接使用进行日志输出。 - 服务注册模块：基于
etcd框架封装的注册模块直接使用，进行聊天消息存储子服
务的注册。 - 数据库数据操作模块：基于
odb-mysql数据管理封装的模块，进行数据库数据操作，用于从MQ中消费到消息后，向数据库中存储一份，以便于通过时间进行范围性查找。
a. 从数据库根据指定用户的所有好友信息 rpc服务模块：基于brpc框架搭建rpc服务器。- 服务发现与调用模块：基于
etcd框架与brpc框架封装的服务发现与调用模块，
a. 连接文件管理子服务：将文件/语音/图片类型的消息以及用户头像之类的文件数据转储到文件管理子服务。
b. 连接用户管理子服务：在消息搜索时，根据发送用户的ID获取发送者用户信息 ES客户端模块：基于elasticsearch框架实现访问客户端，向es服务器进行文本聊天消息的存储，以便于文本消息的关键字搜索。MQ消费模块：基于rabbitmq-client封装的消费者模块从消息队列服务器消费获取聊天消息，将文本消息存储到ElasticSearch服务，将文件消息转储到文件管理子服务，所有消息的简息都需要向数据库存储一份。
模块功能示意图：
接口实现流程
最近 N 条消息获取：
- 从请求中，获取会话
ID， 要获取的消息数量 - 访问数据库，从数据库中按时间排序，获取指定数量的消息简略信息（
消息 ID，会话 ID，消息类型，产生时间，发送者用户ID，文本消息内容，文件消息元信息） - 循环构造完整消息（从用户子服务获取消息的发送者用户信息，从文件子服务获取文件/语音/图片数据）
- 组织响应返回给网关服务器。
指定时间段消息搜索：
- 从请求中，获取会话
ID， 要获取的消息的起始时间与结束时间。 - 访问数据库，从数据库中按时间进行范围查询，获取消息简略信息（消息 ID会
话 ID，消息类型，产生时间，发送者用户 ID，文本消息内容，文件消息元信息） - 循环构造完整消息（从用户子服务获取消息的发送者用户信息，从文件子服务获取文件/语音/图片数据）
- 组织响应返回给网关服务器。
关键字消息搜索：
- 从请求中，获取
会话 ID， 搜索关键字。 - 基于封装的
ES 客户端，访问ES 服务器进行文本消息搜索（以消息内容进行搜索，以会话ID 进行过滤），从 ES 服务器获取到消息简息（消息 ID，会话 ID， 文本消息内容）。 - 循环从数据库根据消息
ID 获取消息简息（消息 ID，消息类型，会话 ID，发送者ID，产生时间，文本消息内容，文件消息元数据）。 - 循环从用户子服务获取所有消息的发送者用户信息，构造完整消息。
- 组织响应返回给网关服务器
mysql数据管理
odb映射文件
message.hxx
#pragma once
#include <string>
#include <cstddef>
#include <odb/nullable.hxx>
#include <odb/core.hxx>
#include <boost/date_time/posix_time/posix_time.hpp>
namespace bite_im
{
#pragma db object table("message")
class Message
{
public:
Message() {}
Message(const std::string &mid,
const std::string &ssid,
const std::string &uid,
const unsigned char mtype,
const boost::posix_time::ptime &ctime) : _message_id(mid), _session_id(ssid),
_user_id(uid), _message_type(mtype),
_create_time(ctime) {}
std::string message_id() const { return _message_id; }
void message_id(const std::string &val) { _message_id = val; }
std::string session_id() const { return _session_id; }
void session_id(const std::string &val) { _session_id = val; }
std::string user_id() const { return _user_id; }
void user_id(const std::string &val) { _user_id = val; }
unsigned char message_type() const { return _message_type; }
void message_type(unsigned char val) { _message_type = val; }
boost::posix_time::ptime create_time() const { return _create_time; }
void create_time(const boost::posix_time::ptime &val) { _create_time = val; }
std::string content() const
{
if (!_content)
return std::string();
return *_content;
}
void content(const std::string &val) { _content = val; }
std::string file_id() const
{
if (!_file_id)
return std::string();
return *_file_id;
}
void file_id(const std::string &val) { _file_id = val; }
std::string file_name() const
{
if (!_file_name)
return std::string();
return *_file_name;
}
void file_name(const std::string &val) { _file_name = val; }
unsigned int file_size() const
{
if (!_file_size)
return 0;
return *_file_size;
}
void file_size(unsigned int val) { _file_size = val; }
private:
friend class odb::access;
#pragma db id auto
unsigned long _id;
#pragma db type("varchar(64)") index unique
std::string _message_id;
#pragma db type("varchar(64)") index
std::string _session_id; // 所属会话ID
#pragma db type("varchar(64)")
std::string _user_id; // 发送者用户ID
unsigned char _message_type; // 消息类型 0-文本;1-图片;2-文件;3-语音
#pragma db type("TIMESTAMP")
boost::posix_time::ptime _create_time; // 消息的产生时间
odb::nullable<std::string> _content; // 文本消息内容--非文本消息可以忽略
#pragma db type("varchar(64)")
odb::nullable<std::string> _file_id; // 文件消息的文件ID -- 文本消息忽略
#pragma db type("varchar(128)")
odb::nullable<std::string> _file_name; // 文件消息的文件名称 -- 只针对文件消息有效
odb::nullable<unsigned int> _file_size; // 文件消息的文件大小 -- 只针对文件消息有效
};
// odb -d mysql --std c++11 --generate-query --generate-schema --profile boost/date-time person.hxx
}
进入到message目录下:
odb -d mysql --std c++11 --generate-query --generate-schema --profile boost/date-time ../odb/message.hxx
#导入到mydsql
mysql -uroot -p <message.sql
然后构造一些假数据:
INSERT `message` VALUES(null,"消息ID1","会话ID1","用户ID1",0,now(),null,null,null,null),
(null,"消息ID2","会话ID2","用户ID2",0,now(),null,null,null,null),
(null,"消息ID3","会话ID3","用户ID3",0,now(),null,null,null,null),
(null,"消息ID4","会话ID4","用户ID4",0,now(),null,null,null,null),
(null,"消息ID5","会话ID5","用户ID5",0,now(),null,null,null,null);
select * from message where session_id='会话ID3' order by create_time desc limit 3;
mysql_message.hpp
#include "mysql.hpp"
#include "message.hxx"
#include "message-odb.hxx"
namespace bite_im
{
class MessageTable
{
public:
using ptr = std::shared_ptr<MessageTable>;
MessageTable(const std::shared_ptr<odb::core::database> &db) : _db(db) {}
~MessageTable() {}
// 添加会话信息
bool insert(Message &msg)
{
try
{
odb::transaction trans(_db->begin());
_db->persist(msg);
trans.commit();
}
catch (std::exception &e)
{
LOG_ERROR("新增消息失败 {}:{}！", msg.message_id(), e.what());
return false;
}
return true;
}
// 删除会话消息
bool remove(const std::string &ssid)
{
try
{
odb::transaction trans(_db->begin());
typedef odb::query<Message> query;
typedef odb::result<Message> result;
_db->erase_query<Message>(query::session_id == ssid);
trans.commit();
}
catch (std::exception &e)
{
LOG_ERROR("删除会话所有消息失败 {}:{}！", ssid, e.what());
return false;
}
return true;
}
// 获取最近n条消息
std::vector<Message> recent(const std::string &ssid, int count)
{
std::vector<Message> res;
try
{
odb::transaction trans(_db->begin());
typedef odb::query<Message> query;
typedef odb::result<Message> result;
// 本次查询是以ssid作为过滤条件，然后进行以时间字段进行逆序，通过limit限制数量
// session_id='xx' order by create_time desc limit count;
std::stringstream cond;
cond << "session_id='" << ssid << "' ";
cond << "order by create_time desc limit " << count;
result r(_db->query<Message>(cond.str()));
for (result::iterator i(r.begin()); i != r.end(); ++i)
{
res.push_back(*i);
}
std::reverse(res.begin(), res.end());
trans.commit();
}
catch (std::exception &e)
{
LOG_ERROR("获取最近消息失败:{}-{}-{}！", ssid, count, e.what());
}
return res;
}
// 获取区间消息
std::vector<Message> range(const std::string &ssid,
boost::posix_time::ptime &stime,
boost::posix_time::ptime &etime)
{
std::vector<Message> res;
try
{
odb::transaction trans(_db->begin());
typedef odb::query<Message> query;
typedef odb::result<Message> result;
// 获取指定会话指定时间段的信息
// 直接使用query内部字段
result r(_db->query<Message>(query::session_id == ssid &&
query::create_time >= stime &&
query::create_time <= etime));
for (result::iterator i(r.begin()); i != r.end(); ++i)
{
res.push_back(*i);
}
trans.commit();
}
catch (std::exception &e)
{
LOG_ERROR("获取区间消息失败:{}-{}:{}-{}！", ssid,
boost::posix_time::to_simple_string(stime),
boost::posix_time::to_simple_string(etime), e.what());
}
return res;
}
private:
std::shared_ptr<odb::core::database> _db;
};
}
main.cc
#include "mysql_message.hpp"
#include <gflags/gflags.h>
DEFINE_bool(run_mode, false, "程序的运行模式，false-调试； true-发布；");
DEFINE_string(log_file, "", "发布模式下，用于指定日志的输出文件");
DEFINE_int32(log_level, 0, "发布模式下，用于指定日志输出等级");
void insert_test(bite_im::MessageTable &tb)
{
bite_im::Message m1("消息ID1", "会话ID1", "用户ID1", 0, boost::posix_time::time_from_string("2002-01-20 23:59:59.000"));
tb.insert(m1);
bite_im::Message m2("消息ID2", "会话ID1", "用户ID2", 0, boost::posix_time::time_from_string("2002-01-21 23:59:59.000"));
tb.insert(m2);
bite_im::Message m3("消息ID3", "会话ID1", "用户ID3", 0, boost::posix_time::time_from_string("2002-01-22 23:59:59.000"));
tb.insert(m3);
bite_im::Message m4("消息ID4", "会话ID2", "用户ID4", 0, boost::posix_time::time_from_string("2002-01-20 23:59:59.000"));
tb.insert(m4);
bite_im::Message m5("消息ID5", "会话ID2", "用户ID5", 0, boost::posix_time::time_from_string("2002-01-21 23:59:59.000"));
tb.insert(m5);
}
void remove_test(bite_im::MessageTable &tb)
{
tb.remove("会话ID2");
}
void recent_test(bite_im::MessageTable &tb)
{
auto res = tb.recent("会话ID1", 2);
auto begin = res.rbegin();
auto end = res.rend();
for (; begin != end; ++begin)
{
std::cout << begin->message_id() << std::endl;
std::cout << begin->session_id() << std::endl;
std::cout << begin->user_id() << std::endl;
std::cout << boost::posix_time::to_simple_string(begin->create_time()) << std::endl;
}
}
void range_test(bite_im::MessageTable &tb)
{
boost::posix_time::ptime stime(boost::posix_time::time_from_string("2002-01-20 23:59:59.000"));
boost::posix_time::ptime etime(boost::posix_time::time_from_string("2002-01-21 23:59:59.000"));
auto res = tb.range("会话ID1", stime, etime);
for (const auto &m : res)
{
std::cout << m.message_id() << std::endl;
std::cout << m.session_id() << std::endl;
std::cout << m.user_id() << std::endl;
std::cout << boost::posix_time::to_simple_string(m.create_time()) << std::endl;
}
}
int main(int argc, char *argv[])
{
google::ParseCommandLineFlags(&argc, &argv, true);
bite_im::init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);
auto db = bite_im::ODBFactory::create("root", "123456", "127.0.0.1", "bite_im", "utf8", 0, 1);
bite_im::MessageTable tb(db);
insert_test(tb);
// remove_test(tb);
// recent_test(tb);
// range_test(tb);
return 0;
}
makefile
CFLAGS = -I/usr/include/mysql -I../../../odb/ -I../../../common -I./
main : main.cc message-odb.cxx
g++ -std=c++17 $^ -o $@ -lodb-mysql $(CFLAGS) -lodb -lodb-boost -lfmt -lspdlog -lgflags
这里我们的消息数据库的操作就封装完毕了！
ES数据管理模块
json.cpp
POST /message/_doc
{
"settings" : {
"analysis" : {
"analyzer" : {
"ik" : {
"tokenizer" : "ik_max_word"
}
}
}
},
"mappings" : {
"dynamic" : true,
"properties" : {
"chat_session_id" : {
"type" : "long",
"analyzer" : "standard"
},
"message_id" : {
"type" : "long",
"enabled":false
},
"user_id" : {
"type" : "long",
"enabled":false
},
"create_time" : {
"type" : "long",
"enabled":false
},
"content" : {
"type" : "text",
"analyzer" : "ik_max_word"
}
}
}
}
GET /message/_doc/_search?pretty
{
"query": {
"match_all": {}
}
}
POST /message/_doc/_bulk
{"index":{"_id":"1"}}
{"chat_session_id" : "会话 ID1","message_id" : "消息ID1","content" : "吃饭了么？ ","user_id":"用户ID1","create_time":12345678910}
{"index":{"_id":"2"}}
{"chat_session_id" : "会话 ID1","message_id" : "消息ID2","content" : "吃的盖浇饭。 ","user_id":"用户ID2","create_time":12345678910}
{"index":{"_id":"3"}}
{"chat_session_id" : "会话 ID2","message_id" : "消息ID3","content" : "昨天吃饭了么？ ","user_id":"用户ID2","create_time":12345678910}
{"index":{"_id":"4"}}
{"chat_session_id" : "会话 ID2","message_id" : "消息ID4","content" : "昨天吃的盖浇饭。 ","user_id":"用户ID2","create_time":12345678910}
过滤搜索
GET /message/_doc/_search?pretty
{
"query" : {
"bool" : {
"must" : [
{
"term" : { "chat_session_id.keyword" : "会话 ID1" }
},
{
"match" : { "content" : "盖浇饭" }
}
]
}
}
}
封装测试:
es_data.hpp
#include "icsearch.hpp"
#include "user.hxx"
#include "message.hxx"
namespace bite_im
{
class ESClientFactory
{
public:
static std::shared_ptr<elasticlient::Client> create(const std::vector<std::string> host_list)
{
return std::make_shared<elasticlient::Client>(host_list);
}
};
// es的操作
class ESUser
{
public:
using ptr = std::shared_ptr<ESUser>;
ESUser(const std::shared_ptr<elasticlient::Client> &client) : _es_client(client) {}
bool createIndex() // 创建索引
{
bool ret = ESIndex(_es_client, "user")
.append("user_id", "keyword", "standard", true)
.append("nickname")
.append("phone", "keyword", "standard", true)
.append("description", "text", "standard", false)
.append("avatar_id", "keyword", "standard", false)
.create();
if (ret == false)
{
LOG_INFO("用户信息索引创建失败!");
return false;
}
LOG_INFO("用户信息索引创建成功!");
return true;
}
// 新增或更新数据
bool appendData(const std::string &uid, // uid
const std::string &phone, // 电话
const std::string &nickname, // 昵称
const std::string &description, // 签名
const std::string &avatar_id) // 头像id
{
bool ret = ESInsert(_es_client, "user")
.append("user_id", uid)
.append("nickname", nickname)
.append("phone", phone)
.append("description", description)
.append("avatar_id", avatar_id)
.insert(uid);
if (ret == false)
{
LOG_ERROR("用户数据插入/更新失败!");
return false;
}
LOG_INFO("用户数据新增/更新成功!");
return true;
}
// 过滤信息
std::vector<User> search(const std::string &key, const std::vector<std::string> &uid_list)
{
std::vector<User> res;
Json::Value json_user = ESSearch(_es_client, "user")
.append_should_match("phone.keyword", key)
.append_should_match("user_id.keyword", key)
.append_should_match("nickname", key)
.append_must_not_terms("user_id.keyword", uid_list)
.search();
if (json_user.isArray() == false)
{
LOG_ERROR("用户搜索结果为空，或者结果不是数组类型");
return res; // 失败返回空的数组
}
int sz = json_user.size();
LOG_DEBUG("检索结果条目数量：{}", sz);
for (int i = 0; i < sz; i++)
{
User user;
user.user_id(json_user[i]["_source"]["user_id"].asString());
user.nickname(json_user[i]["_source"]["nickname"].asString());
user.description(json_user[i]["_source"]["description"].asString());
user.phone(json_user[i]["_source"]["phone"].asString());
user.avatar_id(json_user[i]["_source"]["avatar_id"].asString());
res.push_back(user);
}
return res;
}
private:
// const std::string _uid_key = "user_id";
// const std::string _desc_key = "user_id";
// const std::string _phone_key = "user_id";
// const std::string _name_key = "user_id";
// const std::string _avatar_key = "user_id";
std::shared_ptr<elasticlient::Client> _es_client;
};
class ESMessage
{
public:
using ptr = std::shared_ptr<ESMessage>;
ESMessage(const std::shared_ptr<elasticlient::Client> &es_client) : _es_client(es_client) {}
bool createIndex()
{
bool ret = ESIndex(_es_client, "message")
.append("user_id", "keyword", "standard", false)
.append("message_id", "keyword", "standard", false)
.append("create_time", "long", "standard", false)
.append("chat_session_id", "keyword", "standard", true)
.append("content")
.create();
if (ret == false)
{
LOG_INFO("消息信息索引创建失败!");
return false;
}
LOG_INFO("消息信息索引创建成功!");
return true;
}
bool appendData(const std::string &user_id,
const std::string &message_id,
const long create_time,
const std::string &chat_session_id,
const std::string &content)
{
bool ret = ESInsert(_es_client, "message")
.append("message_id", message_id)
.append("create_time", create_time)
.append("user_id", user_id)
.append("chat_session_id", chat_session_id)
.append("content", content)
.insert(message_id);
if (ret == false)
{
LOG_ERROR("消息数据插入/更新失败!");
return false;
}
LOG_INFO("消息数据新增/更新成功!");
return true;
}
// 删除
bool remove(const std::string &mid)
{
bool ret = ESRemove(_es_client, "message").remove(mid);
if (ret == false)
{
LOG_ERROR("消息数据删除失败!");
return false;
}
LOG_INFO("消息数据删除成功!");
return true;
}
// 搜索
std::vector<bite_im::Message> search(const std::string &key, const std::string &ssid)
{
std::vector<bite_im::Message> res;
Json::Value json_user = ESSearch(_es_client, "message")
.append_must_term("chat_session_id.keyword", ssid)
.append_must_match("content", key)
.search();
if (json_user.isArray() == false)
{
LOG_ERROR("用户搜索结果为空，或者结果不是数组类型");
return res;
}
int sz = json_user.size();
LOG_DEBUG("检索结果条目数量：{}", sz);
for (int i = 0; i < sz; i++)
{
bite_im::Message message;
message.user_id(json_user[i]["_source"]["user_id"].asString());
message.message_id(json_user[i]["_source"]["message_id"].asString());
boost::posix_time::ptime ctime(boost::posix_time::from_time_t(
json_user[i]["_source"]["create_time"].asInt64()));
message.create_time(ctime);
message.session_id(json_user[i]["_source"]["chat_session_id"].asString());
message.content(json_user[i]["_source"]["content"].asString());
res.push_back(message);
}
return res;
}
private:
std::shared_ptr<elasticlient::Client> _es_client;
};
}
main.cc
#include "../../../common/data_es.hpp"
#include <gflags/gflags.h>
DEFINE_bool(run_mode, false, "程序的运行模式，false-调试； true-发布；");
DEFINE_string(log_file, "", "发布模式下，用于指定日志的输出文件");
DEFINE_int32(log_level, 0, "发布模式下，用于指定日志输出等级");
DEFINE_string(es_host, "http://127.0.0.1:9200/", "es服务器URL");
int main(int argc, char *argv[])
{
google::ParseCommandLineFlags(&argc, &argv, true);
bite_im::init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);
auto es_client = bite_im::ESClientFactory::create({FLAGS_es_host});
auto es_msg = std::make_shared<bite_im::ESMessage>(es_client);
es_msg->createIndex();
es_msg->appendData("用户ID1", "消息ID1", 1723025035, "会话ID1", "吃饭了吗？");
es_msg->appendData("用户ID2", "消息ID2", 1723025035 - 100, "会话ID1", "吃的盖浇饭！");
es_msg->appendData("用户ID3", "消息ID3", 1723025035, "会话ID2", "吃饭了吗？");
es_msg->appendData("用户ID4", "消息ID4", 1723025035 - 100, "会话ID2", "吃的盖浇饭！");
//等会数据创立完后，把这个放开
/*auto res = es_msg->search("盖浇", "会话ID1");
for (auto &u : res)
{
std::cout << "-----------------" << std::endl;
std::cout << u.user_id() << std::endl;
std::cout << u.message_id() << std::endl;
std::cout << u.session_id() << std::endl;
std::cout << boost::posix_time::to_simple_string(u.create_time()) << std::endl;
std::cout << u.content() << std::endl;
}*/
return 0;
}
说明我们的常见索引没有问题，接下来查询：
走到这里我们的ES封装是没有问题的！
消息存储子服务的实现
在test目录下:
protoc --cpp_out=./ -I ../../proto --experimental_allow_proto3_optional message.proto
message_server.hpp
// 实现消息存储子服务
#include <brpc/server.h>
#include <butil/logging.h>
#include "data_es.hpp" // es数据管理客户端封装
#include "mysql_message.hpp" // mysql数据管理客户端封装
#include "etcd.hpp" // 服务注册模块封装
#include "logger.hpp" // 日志模块封装
#include "utils.hpp" // 基础工具接口
#include "channel.hpp" // 信道管理模块封装
#include "rabbitmq.hpp"
#include "message.pb.h" // protobuf框架代码
#include "base.pb.h" // protobuf框架代码
#include "file.pb.h" // protobuf框架代码
#include "user.pb.h" // protobuf框架代码
namespace bite_im
{
class MessageServiceImpl : public bite_im::MsgStorageService
{
public:
MessageServiceImpl(
const std::shared_ptr<elasticlient::Client> &es_client,
const std::shared_ptr<odb::core::database> &mysql_client,
const ServiceManager::ptr &channel_manager,
const std::string &file_service_name,
const std::string &user_service_name) : _es_message(std::make_shared<ESMessage>(es_client)),
_mysql_message(std::make_shared<MessageTable>(mysql_client)),
_file_service_name(file_service_name),
_user_service_name(user_service_name),
_mm_channels(channel_manager)
{
_es_message->createIndex();
}
~MessageServiceImpl() {}
// 提取区间消息接口
/////////
virtual void GetHistoryMsg(::google::protobuf::RpcController *controller,
const ::bite_im::GetHistoryMsgReq *request,
::bite_im::GetHistoryMsgRsp *response,
::google::protobuf::Closure *done)
{
brpc::ClosureGuard rpc_guard(done);
auto err_response = [this, response](const std::string &rid,
const std::string &errmsg) -> void
{
response->set_request_id(rid);
response->set_success(false);
response->set_errmsg(errmsg);
return;
};
// 1. 提取关键要素:会话ID,起始时间,结束时间
std::string rid = request->request_id();
std::string chat_ssid = request->chat_session_id();
boost::posix_time::ptime stime = boost::posix_time::from_time_t(request->start_time());
boost::posix_time::ptime etime = boost::posix_time::from_time_t(request->over_time());
// 2. 从数据库中进行消息查询
auto msg_lists = _mysql_message->range(chat_ssid, stime, etime);
if (msg_lists.empty())
{
response->set_request_id(rid);
response->set_success(true);
return;
}
// 3. 统计所有文件类型消息的文件ID，并从文件子服务进行批量文件下载
std::unordered_set<std::string> file_id_lists;
for (const auto &msg : msg_lists)
{
if (msg.file_id().empty()) // 空了
continue;
LOG_DEBUG("需要下载的文件ID： {}", msg.file_id());
file_id_lists.insert(msg.file_id());
}
std::unordered_map<std::string, std::string> file_data_lists;
bool ret = _GetFile(rid, file_id_lists, file_data_lists);
if (ret == false)
{
LOG_ERROR("{} 批量文件数据下载失败！", rid);
return err_response(rid, "批量文件数据下载失败!");
}
// 4. 统计所有消息的发送者用户ID，从用户子服务进行批量用户信息获取
std::unordered_set<std::string> user_id_lists; // {猪爸爸吧， 祝妈妈，猪爸爸吧，祝爸爸}
for (const auto &msg : msg_lists)
{
user_id_lists.insert(msg.user_id());
}
std::unordered_map<std::string, UserInfo> user_lists;
ret = _GetUser(rid, user_id_lists, user_lists);
if (ret == false)
{
LOG_ERROR("{} 批量用户数据获取失败！", rid);
return err_response(rid, "批量用户数据获取失败!");
}
// 5. 组织响应
response->set_request_id(rid);
response->set_success(true);
for (const auto &msg : msg_lists)
{
auto message_info = response->add_msg_list();
message_info->set_message_id(msg.message_id());
message_info->set_chat_session_id(msg.session_id());
message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);
switch (msg.message_type())
{
case MessageType::STRING:
message_info->mutable_message()->set_message_type(MessageType::STRING);
message_info->mutable_message()->mutable_string_message()->set_content(msg.content());
break;
case MessageType::IMAGE:
message_info->mutable_message()->set_message_type(MessageType::IMAGE);
message_info->mutable_message()->mutable_image_message()->set_file_id(msg.file_id());
message_info->mutable_message()->mutable_image_message()->set_image_content(file_data_lists[msg.file_id()]);
break;
case MessageType::FILE:
message_info->mutable_message()->set_message_type(MessageType::FILE);
message_info->mutable_message()->mutable_file_message()->set_file_id(msg.file_id());
message_info->mutable_message()->mutable_file_message()->set_file_size(msg.file_size());
message_info->mutable_message()->mutable_file_message()->set_file_name(msg.file_name());
message_info->mutable_message()->mutable_file_message()->set_file_contents(file_data_lists[msg.file_id()]);
break;
case MessageType::SPEECH:
message_info->mutable_message()->set_message_type(MessageType::SPEECH);
message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.file_id());
message_info->mutable_message()->mutable_speech_message()->set_file_contents(file_data_lists[msg.file_id()]);
break;
default:
LOG_ERROR("消息类型错误！！");
return;
}
}
return;
}
// 获取n条消息
virtual void GetRecentMsg(::google::protobuf::RpcController *controller,
const ::bite_im::GetRecentMsgReq *request,
::bite_im::GetRecentMsgRsp *response,
::google::protobuf::Closure *done)
{
brpc::ClosureGuard rpc_guard(done);
auto err_response = [this, response](const std::string &rid,
const std::string &errmsg) -> void
{
response->set_request_id(rid);
response->set_success(false);
response->set_errmsg(errmsg);
return;
};
// 1. 提取请求中的关键要素：请求ID，会话ID，要获取的消息数量
std::string rid = request->request_id();
std::string chat_ssid = request->chat_session_id();
int msg_count = request->msg_count();
// 2. 从数据库，获取最近的消息元信息
auto msg_lists = _mysql_message->recent(chat_ssid, msg_count);
if (msg_lists.empty())
{
response->set_request_id(rid);
response->set_success(true);
return;
}
// 3. 统计所有消息中文件类型消息的文件ID列表，从文件子服务下载文件
std::unordered_set<std::string> file_id_lists;
for (const auto &msg : msg_lists)
{
if (msg.file_id().empty())
continue;
LOG_DEBUG("需要下载的文件ID: {}", msg.file_id());
file_id_lists.insert(msg.file_id());
}
std::unordered_map<std::string, std::string> file_data_lists;
bool ret = _GetFile(rid, file_id_lists, file_data_lists);
if (ret == false)
{
LOG_ERROR("{} 批量文件数据下载失败！", rid);
return err_response(rid, "批量文件数据下载失败!");
}
// 4. 统计所有消息的发送者用户ID，从用户子服务进行批量用户信息获取
std::unordered_set<std::string> user_id_lists;
for (const auto &msg : msg_lists)
{
user_id_lists.insert(msg.user_id());
}
std::unordered_map<std::string, UserInfo> user_lists;
ret = _GetUser(rid, user_id_lists, user_lists);
if (ret == false)
{
LOG_ERROR("{} 批量用户数据获取失败！", rid);
return err_response(rid, "批量用户数据获取失败!");
}
// 5. 组织响应
response->set_request_id(rid);
response->set_success(true);
for (const auto &msg : msg_lists)
{
auto message_info = response->add_msg_list();
message_info->set_message_id(msg.message_id());
message_info->set_chat_session_id(msg.session_id());
message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);
switch (msg.message_type())
{
case MessageType::STRING:
message_info->mutable_message()->set_message_type(MessageType::STRING);
message_info->mutable_message()->mutable_string_message()->set_content(msg.content());
break;
case MessageType::IMAGE:
message_info->mutable_message()->set_message_type(MessageType::IMAGE);
message_info->mutable_message()->mutable_image_message()->set_file_id(msg.file_id());
message_info->mutable_message()->mutable_image_message()->set_image_content(file_data_lists[msg.file_id()]);
break;
case MessageType::FILE:
message_info->mutable_message()->set_message_type(MessageType::FILE);
message_info->mutable_message()->mutable_file_message()->set_file_id(msg.file_id());
message_info->mutable_message()->mutable_file_message()->set_file_size(msg.file_size());
message_info->mutable_message()->mutable_file_message()->set_file_name(msg.file_name());
message_info->mutable_message()->mutable_file_message()->set_file_contents(file_data_lists[msg.file_id()]);
break;
case MessageType::SPEECH:
message_info->mutable_message()->set_message_type(MessageType::SPEECH);
message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.file_id());
message_info->mutable_message()->mutable_speech_message()->set_file_contents(file_data_lists[msg.file_id()]);
break;
default:
LOG_ERROR("消息类型错误！！");
return;
}
}
return;
}
// 关键字搜索
virtual void MsgSearch(::google::protobuf::RpcController *controller,
const ::bite_im::MsgSearchReq *request,
::bite_im::MsgSearchRsp *response,
::google::protobuf::Closure *done)
{
brpc::ClosureGuard rpc_guard(done);
auto err_response = [this, response](const std::string &rid,
const std::string &errmsg) -> void
{
response->set_request_id(rid);
response->set_success(false);
response->set_errmsg(errmsg);
return;
};
// 关键字的消息搜索--只针对文本消息
// 1. 从请求中提取关键要素：请求ID，会话ID, 关键字
std::string rid = request->request_id();
std::string chat_ssid = request->chat_session_id();
std::string skey = request->search_key();
// 2. 从ES搜索引擎中进行关键字消息搜索，得到消息列表
auto msg_lists = _es_message->search(skey, chat_ssid);
if (msg_lists.empty())
{
response->set_request_id(rid);
response->set_success(true);
return;
}
// 3. 组织所有消息的用户ID，从用户子服务获取用户信息
std::unordered_set<std::string> user_id_lists;
for (const auto &msg : msg_lists)
{
user_id_lists.insert(msg.user_id());
}
std::unordered_map<std::string, UserInfo> user_lists;
bool ret = _GetUser(rid, user_id_lists, user_lists);
if (ret == false)
{
LOG_ERROR("{} 批量用户数据获取失败！", rid);
return err_response(rid, "批量用户数据获取失败!");
}
// 4. 组织响应
response->set_request_id(rid);
response->set_success(true);
for (const auto &msg : msg_lists)
{
auto message_info = response->add_msg_list();
message_info->set_message_id(msg.message_id());
message_info->set_chat_session_id(msg.session_id());
message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);
message_info->mutable_message()->set_message_type(MessageType::STRING);
message_info->mutable_message()->mutable_string_message()->set_content(msg.content());
}
return;
}
// 消息存储接口
void onMessage(const char *body, size_t sz)
{
LOG_DEBUG("收到新消息，进行存储处理！");
// 1. 取出序列化的消息内容，进行反序列化
bite_im::MessageInfo message;
bool ret = message.ParseFromArray(body, sz);
if (ret == false)
{
LOG_ERROR("对消费到的消息进行反序列化失败！");
return;
}
// 2. 根据不同的消息类型进行不同的处理
std::string file_id, file_name, content;
int64_t file_size;
switch (message.message().message_type())
{
// 1. 如果是一个文本类型消息，取元信息存储到ES中
case MessageType::STRING:
content = message.message().string_message().content();
ret = _es_message->appendData(
message.sender().user_id(),
message.message_id(),
message.timestamp(),
message.chat_session_id(),
content);
if (ret == false)
{
LOG_ERROR("文本消息向存储引擎进行存储失败！");
return;
}
break;
// 2. 如果是一个图片/语音/文件消息，则取出数据存储到文件子服务中，并获取文件ID
case MessageType::IMAGE:
{
const auto &msg = message.message().image_message();
ret = _PutFile("", msg.image_content(), msg.image_content().size(), file_id);
if (ret == false)
{
LOG_ERROR("上传图片到文件子服务失败！");
return;
}
}
break;
case MessageType::FILE:
{
const auto &msg = message.message().file_message();
file_name = msg.file_name();
file_size = msg.file_size();
ret = _PutFile(file_name, msg.file_contents(), file_size, file_id);
if (ret == false)
{
LOG_ERROR("上传文件到文件子服务失败！");
return;
}
}
break;
case MessageType::SPEECH:
{
const auto &msg = message.message().speech_message();
ret = _PutFile("", msg.file_contents(), msg.file_contents().size(), file_id);
if (ret == false)
{
LOG_ERROR("上传语音到文件子服务失败！");
return;
}
}
break;
default:
LOG_ERROR("消息类型错误！");
return;
}
// 3. 提取消息的元信息，存储到mysql数据库中
bite_im::Message msg(message.message_id(),
message.chat_session_id(),
message.sender().user_id(),
message.message().message_type(),
boost::posix_time::from_time_t(message.timestamp()));
msg.content(content);
msg.file_id(file_id);
msg.file_name(file_name);
msg.file_size(file_size);
ret = _mysql_message->insert(msg);
if (ret == false)
{
LOG_ERROR("向数据库插入新消息失败！");
return;
}
}
private:
bool _GetUser(const std::string &rid,
const std::unordered_set<std::string> &user_id_lists,
std::unordered_map<std::string, UserInfo> &user_lists)
{
auto channel = _mm_channels->choose(_user_service_name);
if (!channel)
{
LOG_ERROR("{} 没有可供访问的用户子服务节点！", _user_service_name);
return false;
}
UserService_Stub stub(channel.get());
GetMultiUserInfoReq req;
GetMultiUserInfoRsp rsp;
req.set_request_id(rid);
for (const auto &id : user_id_lists)
{
req.add_users_id(id);
}
brpc::Controller cntl;
stub.GetMultiUserInfo(&cntl, &req, &rsp, nullptr);
if (cntl.Failed() == true || rsp.success() == false)
{
LOG_ERROR("用户子服务调用失败：{}！", cntl.ErrorText());
return false;
}
const auto &umap = rsp.users_info();
for (auto it = umap.begin(); it != umap.end(); ++it)
{
user_lists.insert(std::make_pair(it->first, it->second));
}
return true;
}
bool _GetFile(const std::string &rid,
const std::unordered_set<std::string> &file_id_lists,
std::unordered_map<std::string, std::string> &file_data_lists)
{
auto channel = _mm_channels->choose(_file_service_name);
if (!channel)
{
LOG_ERROR("{} 没有可供访问的文件子服务节点！", _file_service_name);
return false;
}
FileService_Stub stub(channel.get());
GetMultiFileReq req;
GetMultiFileRsp rsp;
req.set_request_id(rid);
for (const auto &id : file_id_lists)
{
req.add_file_id_list(id);
}
brpc::Controller cntl;
stub.GetMultiFile(&cntl, &req, &rsp, nullptr);
if (cntl.Failed() == true || rsp.success() == false)
{
LOG_ERROR("文件子服务调用失败：{}！", cntl.ErrorText());
return false;
}
const auto &fmap = rsp.file_data();
for (auto it = fmap.begin(); it != fmap.end(); ++it)
{
file_data_lists.insert(std::make_pair(it->first, it->second.file_content()));
}
return true;
}
bool _PutFile(const std::string &filename,
const std::string &body,
const int64_t fsize,
std::string &file_id)
{
// 实现文件数据的上传
auto channel = _mm_channels->choose(_file_service_name);
if (!channel)
{
LOG_ERROR("{} 没有可供访问的文件子服务节点！", _file_service_name);
return false;
}
FileService_Stub stub(channel.get());
PutSingleFileReq req;
PutSingleFileRsp rsp;
req.mutable_file_data()->set_file_name(filename);
req.mutable_file_data()->set_file_size(fsize);
req.mutable_file_data()->set_file_content(body);
brpc::Controller cntl;
stub.PutSingleFile(&cntl, &req, &rsp, nullptr);
if (cntl.Failed() == true || rsp.success() == false)
{
LOG_ERROR("文件子服务调用失败：{}！", cntl.ErrorText());
return false;
}
file_id = rsp.file_info().file_id();
return true;
}
private:
ESMessage::ptr _es_message;
MessageTable::ptr _mysql_message;
// 这边是rpc调用客户端相关对象
std::string _user_service_name;
std::string _file_service_name;
ServiceManager::ptr _mm_channels;
};
class MessageServer
{
public:
using ptr = std::shared_ptr<MessageServer>;
MessageServer(const MQClient::ptr &mq_client,
const Discovery::ptr service_discoverer,
const Registry::ptr ®_client,
const std::shared_ptr<elasticlient::Client> &es_client,
const std::shared_ptr<odb::core::database> &mysql_client,
const std::shared_ptr<brpc::Server> &server) : _mq_client(mq_client),
_service_discoverer(service_discoverer),
_registry_client(reg_client),
_es_client(es_client),
_mysql_client(mysql_client),
_rpc_server(server) {}
~MessageServer() {}
// 搭建RPC服务器，并启动服务器
void start()
{
_rpc_server->RunUntilAskedToQuit();
}
private:
Discovery::ptr _service_discoverer;
Registry::ptr _registry_client;
MQClient::ptr _mq_client;
std::shared_ptr<elasticlient::Client> _es_client;
std::shared_ptr<odb::core::database> _mysql_client;
std::shared_ptr<brpc::Server> _rpc_server;
};
class MessageServerBuilder
{
public:
// 构造es客户端对象
void make_es_object(const std::vector<std::string> host_list)
{
_es_client = ESClientFactory::create(host_list);
}
// 构造mysql客户端对象
void make_mysql_object(
const std::string &user,
const std::string &pswd,
const std::string &host,
const std::string &db,
const std::string &cset,
int port,
int conn_pool_count)
{
_mysql_client = ODBFactory::create(user, pswd, host, db, cset, port, conn_pool_count);
}
// 用于构造服务发现客户端&信道管理对象
void make_discovery_object(const std::string ®_host,
const std::string &base_service_name,
const std::string &file_service_name,
const std::string &user_service_name)
{
_user_service_name = user_service_name;
_file_service_name = file_service_name;
_mm_channels = std::make_shared<ServiceManager>();
_mm_channels->declared(file_service_name);
_mm_channels->declared(user_service_name);
LOG_DEBUG("设置文件子服务为需添加管理的子服务：{}", file_service_name);
auto put_cb = std::bind(&ServiceManager::onServiceOnline, _mm_channels.get(), std::placeholders::_1, std::placeholders::_2);
auto del_cb = std::bind(&ServiceManager::onServiceOffline, _mm_channels.get(), std::placeholders::_1, std::placeholders::_2);
_service_discoverer = std::make_shared<Discovery>(reg_host, base_service_name, put_cb, del_cb);
}
// 用于构造服务注册客户端对象
void make_registry_object(const std::string ®_host,
const std::string &service_name,
const std::string &access_host)
{
_registry_client = std::make_shared<Registry>(reg_host);
_registry_client->registry(service_name, access_host);
}
// 用于构造消息队列客户端对象
void make_mq_object(const std::string &user,
const std::string &passwd,
const std::string &host,
const std::string &exchange_name,
const std::string &queue_name,
const std::string &binding_key)
{
_exchange_name = exchange_name;
_queue_name = queue_name;
_mq_client = std::make_shared<MQClient>(user, passwd, host);
_mq_client->declareComponents(exchange_name, queue_name, binding_key);
}
void make_rpc_server(uint16_t port, int32_t timeout, uint8_t num_threads)
{
if (!_es_client)
{
LOG_ERROR("还未初始化ES搜索引擎模块！");
abort();
}
if (!_mysql_client)
{
LOG_ERROR("还未初始化Mysql数据库模块！");
abort();
}
if (!_mm_channels)
{
LOG_ERROR("还未初始化信道管理模块！");
abort();
}
_rpc_server = std::make_shared<brpc::Server>();
MessageServiceImpl *msg_service = new MessageServiceImpl(_es_client,
_mysql_client, _mm_channels, _file_service_name, _user_service_name);
int ret = _rpc_server->AddService(msg_service,
brpc::ServiceOwnership::SERVER_OWNS_SERVICE);
if (ret == -1)
{
LOG_ERROR("添加Rpc服务失败！");
abort();
}
brpc::ServerOptions options;
options.idle_timeout_sec = timeout;
options.num_threads = num_threads;
ret = _rpc_server->Start(port, &options);
if (ret == -1)
{
LOG_ERROR("服务启动失败！");
abort();
}
auto callback = std::bind(&MessageServiceImpl::onMessage, msg_service,
std::placeholders::_1, std::placeholders::_2);
_mq_client->consume(_queue_name, callback);
}
// 构造RPC服务器对象
MessageServer::ptr build()
{
if (!_service_discoverer)
{
LOG_ERROR("还未初始化服务发现模块！");
abort();
}
if (!_registry_client)
{
LOG_ERROR("还未初始化服务注册模块！");
abort();
}
if (!_rpc_server)
{
LOG_ERROR("还未初始化RPC服务器模块！");
abort();
}
MessageServer::ptr server = std::make_shared<MessageServer>(
_mq_client, _service_discoverer, _registry_client,
_es_client, _mysql_client, _rpc_server);
return server;
}
private:
Registry::ptr _registry_client;
std::shared_ptr<elasticlient::Client> _es_client;
std::shared_ptr<odb::core::database> _mysql_client;
std::string _user_service_name;
std::string _file_service_name;
ServiceManager::ptr _mm_channels;
Discovery::ptr _service_discoverer;
std::string _exchange_name;
std::string _queue_name;
MQClient::ptr _mq_client;
std::shared_ptr<brpc::Server> _rpc_server;
};
}
message_server.cc
// 主要实现语音识别子服务的服务器的搭建
#include "message_server.hpp"
DEFINE_bool(run_mode, false, "程序的运行模式，false-调试； true-发布；");
DEFINE_string(log_file, "", "发布模式下，用于指定日志的输出文件");
DEFINE_int32(log_level, 0, "发布模式下，用于指定日志输出等级");
DEFINE_string(registry_host, "http://127.0.0.1:2379", "服务注册中心地址");
DEFINE_string(instance_name, "/message_service/instance", "当前实例名称");
DEFINE_string(access_host, "127.0.0.1:10005", "当前实例的外部访问地址");
DEFINE_int32(listen_port, 10005, "Rpc服务器监听端口");
DEFINE_int32(rpc_timeout, -1, "Rpc调用超时时间");
DEFINE_int32(rpc_threads, 1, "Rpc的IO线程数量");
DEFINE_string(base_service, "/service", "服务监控根目录");
DEFINE_string(file_service, "/service/file_service", "文件管理子服务名称");
DEFINE_string(user_service, "/service/user_service", "用户管理子服务名称");
DEFINE_string(es_host, "http://127.0.0.1:9200/", "ES搜索引擎服务器URL");
DEFINE_string(mysql_host, "127.0.0.1", "Mysql服务器访问地址");
DEFINE_string(mysql_user, "root", "Mysql服务器访问用户名");
DEFINE_string(mysql_pswd, "123456", "Mysql服务器访问密码");
DEFINE_string(mysql_db, "bite_im", "Mysql默认库名称");
DEFINE_string(mysql_cset, "utf8", "Mysql客户端字符集");
DEFINE_int32(mysql_port, 0, "Mysql服务器访问端口");
DEFINE_int32(mysql_pool_count, 4, "Mysql连接池最大连接数量");
DEFINE_string(mq_user, "root", "消息队列服务器访问用户名");
DEFINE_string(mq_pswd, "123456", "消息队列服务器访问密码");
DEFINE_string(mq_host, "127.0.0.1:5672", "消息队列服务器访问地址");
DEFINE_string(mq_msg_exchange, "msg_exchange", "持久化消息的发布交换机名称");
DEFINE_string(mq_msg_queue, "msg_queue", "持久化消息的发布队列名称");
DEFINE_string(mq_msg_binding_key, "msg_queue", "持久化消息的发布队列名称");
int main(int argc, char *argv[])
{
google::ParseCommandLineFlags(&argc, &argv, true);
bite_im::init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);
bite_im::MessageServerBuilder msb;
msb.make_mq_object(FLAGS_mq_user, FLAGS_mq_pswd, FLAGS_mq_host,
FLAGS_mq_msg_exchange, FLAGS_mq_msg_queue, FLAGS_mq_msg_binding_key);
msb.make_es_object({FLAGS_es_host});
msb.make_mysql_object(FLAGS_mysql_user, FLAGS_mysql_pswd, FLAGS_mysql_host,
FLAGS_mysql_db, FLAGS_mysql_cset, FLAGS_mysql_port, FLAGS_mysql_pool_count);
msb.make_discovery_object(FLAGS_registry_host, FLAGS_base_service, FLAGS_file_service, FLAGS_user_service);
msb.make_rpc_server(FLAGS_listen_port, FLAGS_rpc_timeout, FLAGS_rpc_threads);
msb.make_registry_object(FLAGS_registry_host, FLAGS_base_service + FLAGS_instance_name, FLAGS_access_host);
auto server = msb.build();
server->start();
return 0;
}
编译运行:
消息存储子服务的联调测试
因为涉及到联调测试，所以我们直接在server目录下新建一个CMakeLists.txt:
cmake_minimum_required(VERSION 3.1.3)
project(all-test)
add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/message)
add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/user)
add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/file)
add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/speech)
add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/transmite)
set(CMAKE_INSTALL_PREFIX ${CMAKE_CURRENT_BINARY_DIR})
然后进行整体的编译与联调测试:
先修改我们消息转发中的transmite_client.cc构造对应的假数据:
#include "etcd.hpp"
#include "channel.hpp"
#include "utils.hpp"
#include <gflags/gflags.h>
#include <gtest/gtest.h>
#include <thread>
#include "transmite.pb.h"
DEFINE_bool(run_mode, false, "程序的运行模式，false-调试； true-发布；");
DEFINE_string(log_file, "", "发布模式下，用于指定日志的输出文件");
DEFINE_int32(log_level, 0, "发布模式下，用于指定日志输出等级");
DEFINE_string(etcd_host, "http://127.0.0.1:2379", "服务注册中心地址");
DEFINE_string(base_service, "/service", "服务监控根目录");
DEFINE_string(transmite_service, "/service/transmite_service", "服务监控根目录");
bite_im::ServiceManager::ptr sm;
void string_message(const std::string &uid, const std::string &sid, const std::string &msg)
{
auto channel = sm->choose(FLAGS_transmite_service);
if (!channel)
{
std::cout << "获取通信信道失败！" << std::endl;
return;
}
bite_im::MsgTransmitService_Stub stub(channel.get());
bite_im::NewMessageReq req;
bite_im::GetTransmitTargetRsp rsp;
req.set_request_id(bite_im::uuid());
req.set_user_id(uid);
req.set_chat_session_id(sid);
req.mutable_message()->set_message_type(bite_im::MessageType::STRING);
req.mutable_message()->mutable_string_message()->set_content(msg);
brpc::Controller cntl;
stub.GetTransmitTarget(&cntl, &req, &rsp, nullptr);
ASSERT_FALSE(cntl.Failed());
ASSERT_TRUE(rsp.success());
}
void image_message(const std::string &uid, const std::string &sid, const std::string &msg)
{
auto channel = sm->choose(FLAGS_transmite_service);
if (!channel)
{
std::cout << "获取通信信道失败！" << std::endl;
return;
}
bite_im::MsgTransmitService_Stub stub(channel.get());
bite_im::NewMessageReq req;
bite_im::GetTransmitTargetRsp rsp;
req.set_request_id(bite_im::uuid());
req.set_user_id(uid);
req.set_chat_session_id(sid);
req.mutable_message()->set_message_type(bite_im::MessageType::IMAGE);
req.mutable_message()->mutable_image_message()->set_image_content(msg);
brpc::Controller cntl;
stub.GetTransmitTarget(&cntl, &req, &rsp, nullptr);
ASSERT_FALSE(cntl.Failed());
ASSERT_TRUE(rsp.success());
}
void speech_message(const std::string &uid, const std::string &sid, const std::string &msg)
{
auto channel = sm->choose(FLAGS_transmite_service);
if (!channel)
{
std::cout << "获取通信信道失败！" << std::endl;
return;
}
bite_im::MsgTransmitService_Stub stub(channel.get());
bite_im::NewMessageReq req;
bite_im::GetTransmitTargetRsp rsp;
req.set_request_id(bite_im::uuid());
req.set_user_id(uid);
req.set_chat_session_id(sid);
req.mutable_message()->set_message_type(bite_im::MessageType::SPEECH);
req.mutable_message()->mutable_speech_message()->set_file_contents(msg);
brpc::Controller cntl;
stub.GetTransmitTarget(&cntl, &req, &rsp, nullptr);
ASSERT_FALSE(cntl.Failed());
ASSERT_TRUE(rsp.success());
}
void file_message(const std::string &uid, const std::string &sid,
const std::string &filename, const std::string &content)
{
auto channel = sm->choose(FLAGS_transmite_service);
if (!channel)
{
std::cout << "获取通信信道失败！" << std::endl;
return;
}
bite_im::MsgTransmitService_Stub stub(channel.get());
bite_im::NewMessageReq req;
bite_im::GetTransmitTargetRsp rsp;
req.set_request_id(bite_im::uuid());
req.set_user_id(uid);
req.set_chat_session_id(sid);
req.mutable_message()->set_message_type(bite_im::MessageType::FILE);
req.mutable_message()->mutable_file_message()->set_file_contents(content);
req.mutable_message()->mutable_file_message()->set_file_name(filename);
req.mutable_message()->mutable_file_message()->set_file_size(content.size());
brpc::Controller cntl;
stub.GetTransmitTarget(&cntl, &req, &rsp, nullptr);
ASSERT_FALSE(cntl.Failed());
ASSERT_TRUE(rsp.success());
}
int main(int argc, char *argv[])
{
google::ParseCommandLineFlags(&argc, &argv, true);
bite_im::init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);
// 1. 先构造Rpc信道管理对象
sm = std::make_shared<bite_im::ServiceManager>();
sm->declared(FLAGS_transmite_service);
auto put_cb = std::bind(&bite_im::ServiceManager::onServiceOnline, sm.get(), std::placeholders::_1, std::placeholders::_2);
auto del_cb = std::bind(&bite_im::ServiceManager::onServiceOffline, sm.get(), std::placeholders::_1, std::placeholders::_2);
// 2. 构造服务发现对象
bite_im::Discovery::ptr dclient = std::make_shared<bite_im::Discovery>(FLAGS_etcd_host, FLAGS_base_service, put_cb, del_cb);
// 3. 通过Rpc信道管理对象，获取提供Echo服务的信道
// 假数据
string_message("a83b-2d93ce70-0000", "会话ID1", "吃饭了吗？");
string_message("1882-74802db0-0001", "会话ID1", "吃的盖浇饭！！");
image_message("a83b-2d93ce70-0000", "会话ID1", "可爱表情图片数据");
speech_message("a83b-2d93ce70-0000", "会话ID1", "动听猪叫声数据");
file_message("a83b-2d93ce70-0000", "会话ID1", "猪爸爸的文件名称", "猪爸爸的文件数据");
return 0;
}
然后编译:
我们先修改我们的头像数据
到这里我们的基本联调是没有问题的！也就是消息存储是没问题的，接下来测试消息的获取。
先构造假数据：
update message set create_time='2026-03-12 10:00:00' where id=12;
update message set create_time='2026-03-13 10:00:00' where id=13;
update message set create_time='2026-03-14 10:00:00' where id=14;
update message set create_time='2026-03-15 10:00:00' where id=15;
update message set create_time='2026-03-16 10:00:00' where id=16;
message_client.cc
#include "etcd.hpp"
#include "channel.hpp"
#include "utils.hpp"
#include <gflags/gflags.h>
#include <gtest/gtest.h>
#include <thread>
#include <boost/date_time/posix_time/posix_time.hpp>
#include "message.pb.h"
#include "base.pb.h"
#include "user.pb.h"
DEFINE_bool(run_mode, false, "程序的运行模式，false-调试； true-发布；");
DEFINE_string(log_file, "", "发布模式下，用于指定日志的输出文件");
DEFINE_int32(log_level, 0, "发布模式下，用于指定日志输出等级");
DEFINE_string(etcd_host, "http://127.0.0.1:2379", "服务注册中心地址");
DEFINE_string(base_service, "/service", "服务监控根目录");
DEFINE_string(message_service, "/service/message_service", "服务监控根目录");
bite_im::ServiceManager::ptr sm;
void range_test(const std::string &ssid,
const boost::posix_time::ptime &stime,
const boost::posix_time::ptime &etime)
{
auto channel = sm->choose(FLAGS_message_service);
if (!channel)
{
std::cout << "获取通信信道失败！" << std::endl;
return;
}
bite_im::MsgStorageService_Stub stub(channel.get());
bite_im::GetHistoryMsgReq req;
bite_im::GetHistoryMsgRsp rsp;
req.set_request_id(bite_im::uuid());
req.set_chat_session_id(ssid);
req.set_start_time(boost::posix_time::to_time_t(stime));
req.set_over_time(boost::posix_time::to_time_t(etime));
brpc::Controller cntl;
stub.GetHistoryMsg(&cntl, &req, &rsp, nullptr);
ASSERT_FALSE(cntl.Failed());
ASSERT_TRUE(rsp.success());
for (int i = 0; i < rsp.msg_list_size(); i++)
{
std::cout << "-----------------------获取区间消息--------------------------\n";
auto msg = rsp.msg_list(i);
std::cout << msg.message_id() << std::endl;
std::cout << msg.chat_session_id() << std::endl;
std::cout << boost::posix_time::to_simple_string(boost::posix_time::from_time_t(msg.timestamp())) << std::endl;
std::cout << msg.sender().user_id() << std::endl;
std::cout << msg.sender().nickname() << std::endl;
std::cout << msg.sender().avatar() << std::endl;
if (msg.message().message_type() == bite_im::MessageType::STRING)
{
std::cout << "文本消息：" << msg.message().string_message().content() << std::endl;
}
else if (msg.message().message_type() == bite_im::MessageType::IMAGE)
{
std::cout << "图片消息：" << msg.message().image_message().image_content() << std::endl;
}
else if (msg.message().message_type() == bite_im::MessageType::FILE)
{
std::cout << "文件消息：" << msg.message().file_message().file_contents() << std::endl;
std::cout << "文件名称：" << msg.message().file_message().file_name() << std::endl;
}
else if (msg.message().message_type() == bite_im::MessageType::SPEECH)
{
std::cout << "语音消息：" << msg.message().speech_message().file_contents() << std::endl;
}
else
{
std::cout << "类型错误！！\n";
}
}
}
void recent_test(const std::string &ssid, int count)
{
auto channel = sm->choose(FLAGS_message_service);
if (!channel)
{
std::cout << "获取通信信道失败！" << std::endl;
return;
}
bite_im::MsgStorageService_Stub stub(channel.get());
bite_im::GetRecentMsgReq req;
bite_im::GetRecentMsgRsp rsp;
req.set_request_id(bite_im::uuid());
req.set_chat_session_id(ssid);
req.set_msg_count(count);
brpc::Controller cntl;
stub.GetRecentMsg(&cntl, &req, &rsp, nullptr);
ASSERT_FALSE(cntl.Failed());
ASSERT_TRUE(rsp.success());
for (int i = 0; i < rsp.msg_list_size(); i++)
{
std::cout << "----------------------获取最近消息---------------------------\n";
auto msg = rsp.msg_list(i);
std::cout << msg.message_id() << std::endl;
std::cout << msg.chat_session_id() << std::endl;
std::cout << boost::posix_time::to_simple_string(boost::posix_time::from_time_t(msg.timestamp())) << std::endl;
std::cout << msg.sender().user_id() << std::endl;
std::cout << msg.sender().nickname() << std::endl;
std::cout << msg.sender().avatar() << std::endl;
if (msg.message().message_type() == bite_im::MessageType::STRING)
{
std::cout << "文本消息：" << msg.message().string_message().content() << std::endl;
}
else if (msg.message().message_type() == bite_im::MessageType::IMAGE)
{
std::cout << "图片消息：" << msg.message().image_message().image_content() << std::endl;
}
else if (msg.message().message_type() == bite_im::MessageType::FILE)
{
std::cout << "文件消息：" << msg.message().file_message().file_contents() << std::endl;
std::cout << "文件名称：" << msg.message().file_message().file_name() << std::endl;
}
else if (msg.message().message_type() == bite_im::MessageType::SPEECH)
{
std::cout << "语音消息：" << msg.message().speech_message().file_contents() << std::endl;
}
else
{
std::cout << "类型错误！！\n";
}
}
}
void search_test(const std::string &ssid, const std::string &key)
{
auto channel = sm->choose(FLAGS_message_service);
if (!channel)
{
std::cout << "获取通信信道失败！" << std::endl;
return;
}
bite_im::MsgStorageService_Stub stub(channel.get());
bite_im::MsgSearchReq req;
bite_im::MsgSearchRsp rsp;
req.set_request_id(bite_im::uuid());
req.set_chat_session_id(ssid);
req.set_search_key(key);
brpc::Controller cntl;
stub.MsgSearch(&cntl, &req, &rsp, nullptr);
ASSERT_FALSE(cntl.Failed());
ASSERT_TRUE(rsp.success());
for (int i = 0; i < rsp.msg_list_size(); i++)
{
std::cout << "----------------------关键字搜索消息---------------------------\n";
auto msg = rsp.msg_list(i);
std::cout << msg.message_id() << std::endl;
std::cout << msg.chat_session_id() << std::endl;
std::cout << boost::posix_time::to_simple_string(boost::posix_time::from_time_t(msg.timestamp())) << std::endl;
std::cout << msg.sender().user_id() << std::endl;
std::cout << msg.sender().nickname() << std::endl;
std::cout << msg.sender().avatar() << std::endl;
if (msg.message().message_type() == bite_im::MessageType::STRING)
{
std::cout << "文本消息：" << msg.message().string_message().content() << std::endl;
}
else if (msg.message().message_type() == bite_im::MessageType::IMAGE)
{
std::cout << "图片消息：" << msg.message().image_message().image_content() << std::endl;
}
else if (msg.message().message_type() == bite_im::MessageType::FILE)
{
std::cout << "文件消息：" << msg.message().file_message().file_contents() << std::endl;
std::cout << "文件名称：" << msg.message().file_message().file_name() << std::endl;
}
else if (msg.message().message_type() == bite_im::MessageType::SPEECH)
{
std::cout << "语音消息：" << msg.message().speech_message().file_contents() << std::endl;
}
else
{
std::cout << "类型错误！！\n";
}
}
}
int main(int argc, char *argv[])
{
google::ParseCommandLineFlags(&argc, &argv, true);
bite_im::init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);
// 1. 先构造Rpc信道管理对象
sm = std::make_shared<bite_im::ServiceManager>();
sm->declared(FLAGS_message_service);
auto put_cb = std::bind(&bite_im::ServiceManager::onServiceOnline, sm.get(), std::placeholders::_1, std::placeholders::_2);
auto del_cb = std::bind(&bite_im::ServiceManager::onServiceOffline, sm.get(), std::placeholders::_1, std::placeholders::_2);
// 2. 构造服务发现对象
bite_im::Discovery::ptr dclient = std::make_shared<bite_im::Discovery>(FLAGS_etcd_host, FLAGS_base_service, put_cb, del_cb);
boost::posix_time::ptime stime(boost::posix_time::time_from_string("2024-08-02 00:00:00"));
boost::posix_time::ptime etime(boost::posix_time::time_from_string("2024-08-09 00:00:00"));
range_test("会话ID1", stime, etime);
recent_test("会话ID1", 2);
search_test("会话ID1", "盖浇");
return 0;
}
编译运行：
到这里我们的消息存储子服务已经全部测试完毕！！
更多推荐
8
0- 0
已为社区贡献8条内容
所有评论(0)