sonic消息传递机制与架构1

13次阅读

共计 6286 个字符，预计需要花费 16 分钟才能阅读完成。 

sonic 是一个网络操作系统，采用了大量的相互独立的第三方开源组件，这些组件在依赖，编译环境，库，配置方式都有很大的不同。为了让这些组件在 sonic 中相互协作，互不干扰，同时尽量不修改第三方组件的代码，sonic 采用容器技术为各个组件提供独立的运行环境，通过容器间共享网络命名空间进行通信。

​ 各个第三组件有各自的配置文件格式和消息格式，如何让这些组件互通信息了。sonic 采用 redis 数据库作为消息传递平台，通过纯字符消息方式屏蔽各个组件的插件，通过胶水代码将其粘起来。

sonic 消息框架图

实现

sonic 通过 redis 数据库的 发布 - 订阅 机制和 键空间事件 机制实现了整个消息传递机制。

基类

class TableBase {
public:
    TableBase(int dbId, const std::string &tableName)
        : m_tableName(tableName), m_dbId(dbId)
    {
        /* Look up table separator for the provided DB */
        auto it = tableNameSeparatorMap.find(dbId);

        if (it != tableNameSeparatorMap.end())
        {m_tableSeparator = it->second;}
        else
        {SWSS_LOG_NOTICE("Unrecognized database ID. Using default table name separator ('%s')", TABLE_NAME_SEPARATOR_VBAR.c_str());
            m_tableSeparator = TABLE_NAME_SEPARATOR_VBAR;
        }
    }

    std::string getTableName() const { return m_tableName;}
    int getDbId() const { return m_dbId;}

    /* Return the actual key name as a combination of tableName<table_separator>key */
    std::string getKeyName(const std::string &key)
    {if (key == "") return m_tableName;
        else return m_tableName + m_tableSeparator + key;
    }

    /* Return the table name separator being used */
    std::string getTableNameSeparator() const
    {return m_tableSeparator;}

    std::string getChannelName() { return m_tableName + "_CHANNEL";}
private:
    static const std::string TABLE_NAME_SEPARATOR_COLON;
    static const std::string TABLE_NAME_SEPARATOR_VBAR;
    static const TableNameSeparatorMap tableNameSeparatorMap;

    std::string m_tableName;
    std::string m_tableSeparator;
    int m_dbId;
};
class TableEntryWritable {
public:
    virtual ~TableEntryWritable() = default;

    /* Set an entry in the table */
    virtual void set(const std::string &key,
                     const std::vector<FieldValueTuple> &values,
                     const std::string &op = "",
                     const std::string &prefix = EMPTY_PREFIX) = 0;
    /* Delete an entry in the table */
    virtual void del(const std::string &key,
                     const std::string &op = "",
                     const std::string &prefix = EMPTY_PREFIX) = 0;

};

消费者基类

 消费者响应生产者的事件，可以采用阻塞或者轮询的方式处理。sonic 采用了异步事件通知机制 (poll) 进行处理。消费者类必须实现事件通知机制相关的接口。
RedisSelect

该类对异步通知机制 Selectable(select，poll 等)进行了封装，集成该类的派生类可以加入异步事件机制，通过集成该类，消费者可以持续监听事件。

class RedisSelect : public Selectable
{
public:
    /* The database is already alive and kicking, no need for more than a second */
    static constexpr unsigned int SUBSCRIBE_TIMEOUT = 1000;

    RedisSelect(int pri = 0);// 调度优先级

    int getFd() override;
    void readData() override;
    bool hasCachedData() override;
    bool initializedWithData() override;
    void updateAfterRead() override;

    /* Create a new redisContext, SELECT DB and SUBSCRIBE */
    void subscribe(DBConnector* db, const std::string &channelName);

    /* PSUBSCRIBE */
    void psubscribe(DBConnector* db, const std::string &channelName);

    void setQueueLength(long long int queueLength);

protected:
    std::unique_ptr<DBConnector> m_subscribe;
    long long int m_queueLength;// 接收的应答的个数，一个请求一个应答。};
getFd
int RedisSelect::getFd()
{return m_subscribe->getContext()->fd;
}
readData
void RedisSelect::readData()
{
    redisReply *reply = nullptr;

    if (redisGetReply(m_subscribe->getContext(), reinterpret_cast<void**>(&reply)) != REDIS_OK)
        throw std::runtime_error("Unable to read redis reply");

    freeReplyObject(reply);
    m_queueLength++;// 事件加一次，reply = nullptr;
    int status;
    do
    {status = redisGetReplyFromReader(m_subscribe->getContext(), reinterpret_cast<void**>(&reply));
        if(reply != nullptr && status == REDIS_OK)
        {// 一个响应加一次，该值会大于最终处理的循环次数，造成空转，但是不加的话，极端情况下会造成丢失信息问题
            m_queueLength++;
            freeReplyObject(reply);
        }
    }
    while(reply != nullptr && status == REDIS_OK);

    if (status != REDIS_OK)
    {throw std::runtime_error("Unable to read redis reply");
    }
}
hasCachedData
bool RedisSelect::hasCachedData()
{// 判断是否还有消息，存在消息的话，加入 m_ready，保证已经读出来的消息能被处理
    return m_queueLength > 1;
}
updateAfterRead
void RedisSelect::updateAfterRead()
{m_queueLength--;// 假设一次处理一个应答，这里减去 1，即使一次处理了多个消息，依然只减掉 1，造成空转的根本原因}
setQueueLength
void RedisSelect::setQueueLength(long long int queueLength)
{m_queueLength = queueLength;// 设置消息个数，用于构造函数}
subscribe and psubscribe
/* Create a new redisContext, SELECT DB and SUBSCRIBE */
void RedisSelect::subscribe(DBConnector* db, const std::string &channelName)
{m_subscribe.reset(db->newConnector(SUBSCRIBE_TIMEOUT));

    /* Send SUBSCRIBE #channel command */
    std::string s("SUBSCRIBE");
    s += channelName;
    RedisReply r(m_subscribe.get(), s, REDIS_REPLY_ARRAY);
}

/* PSUBSCRIBE */
void RedisSelect::psubscribe(DBConnector* db, const std::string &channelName)
{m_subscribe.reset(db->newConnector(SUBSCRIBE_TIMEOUT));

    /*
     * Send PSUBSCRIBE #channel command on the
     * non-blocking subscriber DBConnector
     */
    std::string s("PSUBSCRIBE");
    s += channelName;
    RedisReply r(m_subscribe.get(), s, REDIS_REPLY_ARRAY);
}

消费者进一步封装基类

class TableEntryPoppable {
public:
    virtual ~TableEntryPoppable() = default;

    /* Pop an action (set or del) on the table */
    virtual void pop(KeyOpFieldsValuesTuple &kco, const std::string &prefix = EMPTY_PREFIX) = 0;

    /* Get multiple pop elements */
    virtual void pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const std::string &prefix = EMPTY_PREFIX) = 0;
};

class TableConsumable : public TableBase, public TableEntryPoppable, public RedisSelect {
public:
    /* The default value of pop batch size is 128 */
    static constexpr int DEFAULT_POP_BATCH_SIZE = 128;// 一次消费 128 条消息

    TableConsumable(int dbId, const std::string &tableName, int pri) : TableBase(dbId, tableName), RedisSelect(pri) {}};

redis lua 执行脚本

EVAL script numkeys key [key ...] arg [arg ...]
首先大家一定要知道 eval 的语法格式，其中：<1> script：你的 lua 脚本
   <2> numkeys:  key 的个数
   <3> key:         redis 中各种数据结构的替代符号
   <4> arg:         你的自定义参数
ok，可能乍一看模板不是特别清楚，下面我可以用官网的小案例演示一下:
eval "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 2 username age jack 20

上面这一串代码大概是什么意思呢？第一个参数的字符串就是 script，也就是 lua 脚本。2 表示 keys 的个数，KEYS[1] 就是 username 的占位符, KEYS[2]就是 age 的占位符，ARGV[1]就是 jack 的占位符，ARGV[2]就是 20 的占位符，，以此类推，，，所以最后的结果应该就是:{return username age jack 20} 是不是有点像 C# 中的占位符:{0}呢？下面我在 Redis 中给大家演示一下：

admin@admin:~$ redis-cli
127.0.0.1:6379> eval "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 2 username age jack 20
1) "username"
2) "age"
3) "jack"
4) "20"
127.0.0.1:6379>

然后我们通过下面命令执行，这种方式和前面介绍的不一样，参数 –eval script key1 key2 , arg1 age2 这种模式，key 和 value 用一个逗号隔开就好了，最后我们也看到了，数据都出来了，对吧.

admin@admin:~$  redis-cli --eval t.lua username age , jack 20                 
1) "username"
2) "age"
3) "jack"
4) "20"
admin@admin:~$
注意上面的逗号左右两边都有空格
  • 脚本也可以在 REPL 模式上执行，不过需要先加载脚本：
admin@admin:~$ redis-cli script load "$(cat t.lua)"                          
"a42059b356c875f0717db19a51f6aaca9ae659ea"
admin@admin:~$
admin@admin:~$ redis-cli
127.0.0.1:6379> EVALSHA a42059b356c875f0717db19a51f6aaca9ae659ea 2 username age jack 20
1) "username"
2) "age"
3) "jack"
4) "20"
127.0.0.1:6379>
  • lua 脚本比较大小，需要使用函数 tonumber 将字符转换成数字，然后比较大小
admin@admin:~$ cat flashsale.lua
local buyNum = ARGV[1]                        -- 本次购买的数量
local goodsKey = KEYS[1]                      -- 本次购买的商品名
local goodsNum = redis.call('get',goodsKey)   -- 获取商品的库存个数
if tonumber(goodsNum) >= tonumber(buyNum)     -- 库存足够，那么出货
    then redis.call('decrby',goodsKey,buyNum) -- 减少本次买的量
    return buyNum                             -- 返回购买的量
else
    return '0'                                -- 数量不够，直接返回 0
end
admin@admin:~$
admin@admin:~$ redis-cli --eval flashsale.lua "pets" , 8
"8"
admin@admin:~$
上面脚本实现的是一个简单的秒杀程序

正文完
c c++
发表至：无分类
2019-07-27
 0