Mar2

【原创】PHP RabbitMQ消息队列应用高级教程

Author: leeon  Click: 1367   Comments: 0 Category: 架构  Tag: rabbitmq,php,amqp

    最近在看rabbitmq的设计哲学和相关概念,有很多需要总结的学习心得可以分享一下。相比以往使用的轻量级高性能beanstalkd队列服务组件,rabbitmq提供了更更丰富的功能,当然他也是可以做到数据持久化的。

    参考了不少资料,但是在PHP下使用rabbitmq的案例并不多,有些代码也略显陈旧,今天主要从一下几个方面开始介绍和总结。

 1. php-amqp 扩展安装

 2. exchange,queue,routing_key, name概念的相关梳理

 3. php-amqp相关API解析

 4. rabbitmq 持久化使用

   

php-amqp扩展安装

    rabbitmq对应php的api其实有好几个,从性能上当然首选pecl的最佳。php-amqp是抽象的消息队列扩展,并不是只针对rabbitmq提供访问服务,只要是支持AMQP协议的消息队列服务程序,都能通过php-amqp插件获得访问能力。
    在安装php-amqp扩展的时候需要先安装了librabbitmq依赖,如果你正在使用centos的系统,我的建议是最好不要使用yum仓库中提供的librabbitmq包,最好手工去下载最新版本的文件进行手工编译安装,因为楼主在用yum安装了ibrabbitmq-devel的时候发现了兼容性问题,导致php始终连接不上rabbitmq。
    官方源文件rabbit-c的安装地址在https://github.com/alanxz/rabbitmq-c 这里,git下来后记得在configure的时候制定prefix安装路径,这样在pecl安装amqp扩展的时候才能让php-amqp包识别到librabbitmq的存在,如果直接configure的话会导致找不到librabbitmq相关文件。
  rabbitmq-安装成功后,我们假定安装路径为/usr/local/librabbitmq ,那么接着安装php-amqp的时候步骤如下:
pecl install amqp
此时提示如下信息时:Set the path to librabbitmq install prefix [autodetect] : 
输入刚才指定的安装路径/usr/local/librabbitmq 回车即可。
编译安装完成后将extension=amqp.so 放入php.ini文件中。

注意在使用rabbitmq的时候默认的账号是guest,但是这个账号限定只能本地访问操控,因此在我们使用rabbitmq远程访问的时候最好新建独立的账号,并进行相应授权,这个操作使用rabbitmqctl命令来控制。

exchange,queue,routing key, name概念的相关梳理

exchange相当于生产者的接收者,同时也相当于接收者数据派送者,类似于一个消息中转站。通常exchange的出现需要和queue配对,并且需要事先申明。exchange类型又主要分为三类:direct(定向直接投递),fanout(广播投递),topic(模糊主题投递)。exchange在申明创建的时候可以定义一个name名字,也可以不定义,不定义的时候rabbitmq会默认赋值一个固定的name。同时对exchange的属性也做了定义:AMQP_DURABLE(持久化,rabbitmq重启后exchange依然存在)和AMQP_PASSIVE(不自动创建exchange,而是检查exchange是否存在) 。

当我们需要检查一个name为Test的exchange是否存在的时候我们可以使用如下代码:
[code="php"]
$connection = new AMQPConnection();
$connection->setHost('127.0.0.1');
$connection->setLogin('leeon');
$connection->setPassword('leeon');
$connection->connect();
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName('Test'); // 设定要检查的exchange的名字
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_PASSIVE); // 使用PASSIVE参数
$exchange->declareExchange();
[/code]

如果Test存在则不会抛出任何异常,如果不存在则会抛出异常
Fatal error: Uncaught AMQPExchangeException: Server channel error: 404, message: NOT_FOUND - no exchange 'Test' in vhost '/'

AMQP_DURABLE参数的设定将会保证rabbitmq的重启不会导致申明定义的exchange丢失。例如我们正常创建了一个exchange后可以在rabbitmq的web控制端看到此状态:


queue的定义主要在消费者这里做定义,定义queue的时候也可以定义一个name,这个name仅仅是对队列名字的命名,不同队列的名字name是不一样的。同样queue的属性定义也有好几个:
AMQP_DURABLE (持久化)
AMQP_PASSIVE (不自动创建queue,而是检查queue是否存在,不存在抛出异常)
AMQP_EXCLUSIVE (排他队列,以此定义命名的队列只有一个)
AMQP_AUTODELETE(当queue释放的时候是否自动删除这个队列在rabbitmq中的记录)

routing key 这个好比一个协商key,当发布者和消费者有协商一致的key策略的时候,消费者就可以合法从生产者手中获取数据。这个routing key主要当exchange设定为direct和topic模式的时候使用,fanout模式不使用routing key,那么这三个模式又有何区别呢?我们来看看如下三个图:




这三个图分别展示了direct,fanout,topic的消息投递规则。

php-amqp相关API解析

php-amqp在最近的版本中改动比较大,例如对于exchange和queue的delcare方法做了改变,修改为了declareExchange和declareQueue。因此网上很多相关的PHP教程事例代码并不能直接拿来学习了。
这里我们主要对publish,consume两个api方法进行讲解。
publish方法:
[code="php"]
public function publish(
$message,
$routing_key = null,
$flags = AMQP_NOPARAM,
array $attributes = array()
) {
}
[/code]
publish方法在生产者的代码逻辑中使用,message参数是需要传递的具体消息内容,这里我们可以通过序列化或者json字符串的形式进行复杂数据的封装。routing_key参数就是前文所说的exchange和queue之间的路由协商key。当我们使用fanout模式的时候,routing_key定义为null。 flags参数包含了两个定义:AMQP_MANDATORY 和 AMQP_IMMEDIATE. 这两个参数在当前php-amqp(1.8.0)版本中并没有实际的作用,因为php-amqp里面的实现调用的是异步模式,并不会等待结果的返回。
[code="cpp"]
/* NOTE: basic.publish is asynchronous and thus will not indicate failure if something goes wrong on the broker */
int status = amqp_basic_publish(
channel_resource->connection_resource->connection_state,
channel_resource->channel_id,
(Z_TYPE_P(exchange_name) == IS_STRING && Z_STRLEN_P(exchange_name) > 0 ? amqp_cstring_bytes(Z_STRVAL_P(exchange_name)) : amqp_empty_bytes), /* exchange */
(key_len > 0 ? amqp_cstring_bytes(key_name) : amqp_empty_bytes), /* routing key */
(AMQP_MANDATORY & flags) ? 1 : 0, /* mandatory */
(AMQP_IMMEDIATE & flags) ? 1 : 0, /* immediate */
&props,
php_amqp_long_string(msg, msg_len) /* message body */
);
[/code]

我们重点需要关注的是attributes的参数设定,这里是一个数组,可以同时定义多个属性,包括如下信息:
content_type: MIME类型定义,例如gzip
message_id: 消息唯一识别id
user_id: 用来记录这个消息的发送者唯一识别id
app_id: 用来记录这个消息的生产者的唯一识别id
delivery_mode: 消息的传送模式 1代表非持久态 2代表持久态(如果消息需要高可用,例如rabbitmq宕机后数据不丢失,name这里就要设置成2)
priority: 消息的优先级,设定值从0到9
timestamp: 消息当前被发送的时间的时间戳
expiration: 消息的过期时间,这个时间单位是毫秒
type: 消息的类型名字
reply_to: 通常用在RPC中,告知需要回复的队列
headers: 消息头部数据表,可以定义多个自定义数据。

consume方法:
[code="php"]
public function consume(
callable $callback = null,
$flags = AMQP_NOPARAM,
$consumerTag = null
) {
}
[/code]

consume方法需要传递一个回调函数,这里不支持以前我们php中的惯用法,传递一个方法名,而是需要传递一个匿名含函数的变量值。flags参数可选AMQP_AUTOACK设定,如果设定AMQP_AUTOACK则当消费者接收到消息后向rabbitmq自动确认消息接收成功,如果不设定AMQP_AUTOACK,则需要我们在代码中调用ack方法来确认消息接收状态。例如如下一段代码:
[code="php"]
$connection = new AMQPConnection();
$connection->setHost('127.0.0.1');
$connection->setLogin('leeon');
$connection->setPassword('leeon');
$connection->connect();

$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);

$queue = new AMQPQueue($channel);
$queue->setName('GGGGGG');
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->declareQueue();

$queue->bind('topic','aaa.#');
$callback = function (AMQPEnvelope $message, AMQPQueue $queue) {
echo $message->getBody() . PHP_EOL;
$queue->ack($message->getDeliveryTag());
};

$queue->consume($callback);
[/code]

     当我们在编写消费者代码的时候,应当都需要调用bind方法来对exchange进行绑定,注意exchange的name一定要已经创建,否则调用bind方法会抛出异常。当使用topic和direct模型时,则需要制定明确的routing key。所以为什么大家在网上看到的PHP AMQP示例代码中,对于生产者和消费者都会申明一次exchange的缘由了,因为PHP的无状态性,为了保证程序代码的完整性,我们需要养成一个习惯,在生产者和消费者中都申明一次exchange和queue。
    为何也要同时申明queue呢?因为消息的投递一定要有明确的queue队列来接收,当我们使用publish来发布后如果找不到合法的queue,这条消息就被rabbitmq丢弃掉了。

rabbitmq 持久化使用

rabbitmq的持久化分为三个部分:exchange,queue 和消息体。当需要一个高可用的消息队列服务时,我们需要同时对exchange和queue在创建的时候申明durable状态,在生产者发送的消息中定义delivery_mode为2即可。例如如下代码:
[code="php"]
$connection = new AMQPConnection();
$connection->setHost('127.0.0.1');
$connection->setLogin('leeon');
$connection->setPassword('leeon');
$connection->connect();
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName('ex');
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange();

try {

$queue = new AMQPQueue($channel);
$queue->setName('AAA');
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();
$queue->bind('ex','keykey');

var_dump( $exchange->publish(
'hello world!',
'keykey',
AMQP_NOPARAM,
[
'delivery_mode'=>2
]
));

}catch (Exception $e){
var_dump($e);
}
[/code]

Feb14

【转载】斗鱼直播平台后端RPC架构浅析

Author: 陈厚道  Click: 812   Comments: 0 Category: 架构  Tag: douyu,斗鱼,rpc

# 背景

# 关键设计点

## 模块化

## 资源隔离

## 权限控制

### RPC框架的需求分析和概要设计

#### 发展与现状

- RPC框架指的是能够完成RPC调用的解决方案,除了点对点的RPC协议的具体实现之外,还可以包含服务的发现与注销,提供服务的多台Server的负载均衡、服务的高可用等更多的功能,目前的RPC框架大致有两种不同的侧重方向,一种偏重于服务治理,另一种偏重于跨语言调用。

- 开源的RPC框架介绍:Dubbo、DubboX、Thrift、Motan。其中Dubbo,Dubbox,motan是java生态下的偏向服务治理的RPC框架,Thrift是偏重于跨语言的调用的RPC。

#### RPC框架提供的主要功能

- 服务发现:服务发布、订阅、通知

- 负载均衡:支持一致性Hash、随机请求、轮询、最少连接数优先、低并发优先等分发原则

- 高可用策略:失败重试(FailOver)、快速失败(FailFast)、异常隔离(Server连续失败超过指定次数置为不可用,然后定期进行心跳探测)

- 其他: 调用统计、权限控制、安全、调用链追踪、日志

#### DY RPC框架交互流程1

DY RPC框架中有服务提供方 RPC Server,服务调用方RPC Client,服务注册中心MessageServer三个角色。该框架的RPCServer主要现在用c++写的服务,RPC Client包括php或者RPCServer。

1. RPC Server向MessageServer集群的某个节点B注册服务,并保持长连接。该MessageServe r B节点会通知集群的所有节点。

同时MessageServer B节点也会定时把注册到该节点的RPCServer的服务配置信息同步到 MessageServer集群。

2. RPCClient会连接到MessageServer集群的某个节点A,发起RPC调用。MessageServer A节点会根据RPC调用的参数(服务提供方的ID,GroupID、负载均衡策略等)选择一条合适的

RPC调用链路,比如RPCClient->MessageServerA->MessageServerB->RPCServer,最终到达某个RPCServer,进行函数调用。其中一个RPC调用最多会经过2个MessageServer节点,最少会经过1个MessageServer节点。

3. 当某个RPC Server发生变更时,通过广播的方式,MessageServer集群的所有节点也能比较实时的感知到某个RPCServer发生变更。

TODO RPC流程交互图

#### DY RPC功能模块划分

1. MessageServer在RPC框架这个功能上应该提供的功能,包括服务的注册和发现模块、协议序列化模块、心跳检测模块、负载均衡算法模块,RPC路由模块、失败重试策略模块、超时丢弃策略模块、消息持久化模块。

2. RPCServer要包含RPC治理的组件,主要功能包括RPC的统计、RPC的频率控制、RPC的安全性控制。

##### RPCServer可用性检测模块

每个服务默认都要实现一个类似Ping Pong的 Request和Response,用来给直连RPCServer的MessageServer探测RPCServer是否可用提供依据。不能简单的依赖心跳消息来探测RPCServer是否可用。

##### 负载均衡模块

MessageServer把RPC请求转发给RPCServer Group时,需要支持的负载均衡算法:

1. 随机法(已实现)

2. 轮询法(已实现)目前在生产环境用的这种算法,负载较不均衡。

3. 组内Hash法(已实现)

4. TODO 最少连接法 (最靠谱的负载均衡做法)

斗鱼采用的这些负载均衡算法可以参考这篇微信文章的介绍:http://mp.weixin.qq.com/s/PAOvmzraVlAMECL-PZs2Pg

看服务器响应自己请求的速度就可以判断应该把下一个请求发到哪个服务器端。

具体说是选择活动请求(已经发出去的请求收到响应)数目最少的那个服务端。 只要根据自己以往的调用情况就能做出判断。

5. TODO: 目前的消息系统只支持点到点、点到组。目前还暂不支持点到组内的某个节点的负载均衡算法。

##### 失败重试策略模块

在RPCClient直连的MessageServer上实现RPC失败重试的策略。

- 只有幂等的RPC调用才能重试。

##### 超时丢弃策略模块

在RPCServer的业务层实现超时丢弃的策略,应用场景:发送火箭超时时,客户端提示发送失败,其实是在鱼翅交易服务器出现性能抖动导致。最后的结果就是鱼翅服务器扣除了鱼翅,但是客户端提示发送火箭失败,比较严重的情况是,用户以为提示失败时不会消耗鱼翅,所以不断重新发送火箭。

针对这种类型的RPC,RPCServer的业务层可以根据RPC的配置规则+RPC发起时间来决定是否直接丢弃该RPC。

##### 消息持久化模块

- 在调用RPC时如果指定可达时,才触发消息持久化的机制。

- 因为RPC的调用链最多需要经过4个节点(RPCClient->MessageServerA->MessageServerB->RPCServer),导致RPC不可达到的情况较为复杂,如果采用自研的方案做消息持久化的话,我们可以假设MessageServer的集群比较稳定,RPCServer较不稳定,所以我们持久化的方案是在和RPCServer直连的MessageServer上实现。

- MessageServer上做持久化具体设计要点:

- 正常流程:

- MessageServer将RPC请求转化为消息,以RPCServer的模块id为Key,将消息存入Redis的队列,我们将这个消息称为MessageData;

- 将RPC请求的MessageID作为Key,Value作为保留字段设计,存入Redis的String,我们将这个数据称为MetaData,同时设置这个Key的过期时间为10分钟(暂定),这个操作和上面的操作作为Redis的一个事务来执行;

- 执行完上面的事务后,直接调用RPC的Response,返回给RPCClient;

- RPCServer集群的某个节点从Redis队列取出MessageData,执行RPCHandler。

- 异常流程1:

- 如果在执行RPCHandler的过程中,RPCServer异常,就只会影响一条MessageData。可以通过一些辅助脚本来做补单,考虑一种策略来实现自动化的补单。

- 异常流程2:

- MessageServerA->MessageServerB网络抖动 或者 MessageServerB->Redis的网络抖动都会导致MessageData不能进入队列;

- 在和RPCClient直连的MessageServerA一段时间(先暂定10s)没有收到RPCResponse,就会触发重试机制,重试的上限次数暂定20次,确保整体重试的时间小于MetaData过期的时间就可以,重试流程进入到MessageServerB节点时,如果是重试RPC,查找Redis队列是否有这个MessageData,如果不存在,则执行正常流程。如果存在,则丢弃本次重试,说明上一次重试已经成功了。

##### 增加RPC追踪链日志

- 在RPCClient直连的个MessageServer上给RPC请求赋予一个Global的RPCID;

- RPCID可以从IDMakerServer集群获得,通过一次获得一批ID来获得良好的性能;

- 在RPC经过的每个节点,都需要有规划统一的格式,并上报给大数据平台;

- 在大数据后台,可以根据RPCID查找整个RPC调用链上的信息。

##### RPC治理组件

- RPC调用统计:每个RPC入口增加统计信息,当rpc进入内部业务函数后也有一次统计,统计信息汇入大数据实时统一日志

- RPC频率控制:某个时间单位内,RPC调用不得超过某个数量;如果有超过,则报警。在频率控制粒度方面,采取如下控制策略和监控策略。

- 每个服务的所有RPC在单位时间内的调用频率控制,超出则报警;

- 某个RPC在单位时间内的调用频率控制,超出则报警;

- 定时统计每个Client来源在每个RPC的调用次数,并按照统一格式上传给大数据平台,大数据平台提供按照Client来源、时间查找RPC调用次数的Top 10的类似功能;

- 大数据平台定时对比RPC的历史调用次数和当前调用次数,超过一定的比例就报警。

- RPC安全策略:

- 可以随时关闭某个RPC、某个服务的所有RPC的安全策略;

- ip验证:给一个ip白名单,这个白名单才能发起RPC调用,不建议按照每个RPC调用单独设置ip白名单

- 口令验证:针对某个RPC、某个服务单独设置密码,对大都数服务都设置成统一的密码,不建议针对每个RPC或者每个服务都单独设置密码。因为除了密码,还有一个摘要认证加密算法才能破解RPC的协议。现在密码是运维维护的,摘要认证加密算法是开发维护的。所以不建议对密码的粒度控制得过细

- RPC手动降级:可以随时关闭某个RPC;也可以根据Client来源关闭某个RPC,但对其他Client来源是开启的。

- RPC自动降级: TODO

- 配置文件格式:参考详细设计文档 by 李明

#### 关键数据结构

1. 服务注册协议

```

struct MC_MsgLoginReqNew : public MessageRoot

{

uint8_t  _version;

DWORD    _uid;

char      _user_name[33];

char      _password[33];      //之前的口令字段依旧不使用

int      _module_id          //模块id

};

```

2.RPCClient Request基本结构,同样包括GroupRPCReq(组内随机调用),GroupRPCReq2(组内hash调用)的

```

struct MC_RPC_Req_New : public MessageRoot< MCT_RPC_Req_New, MC_RPC_Req_New >

{

uint8_t _version;        // 版本号

int64 _rpc_global_id;    // 每次调用需要从idMakderServer获得唯一id,RPC追踪链需要依赖该id来识别

int _rpc_option;          // 包括RPC可达,重试,超时丢失等标志,不可叠加

int32 _user_data;          // 自定义用户数据

int _rpc_retry_count;      // 0表示第一次请求,每重试一次+1

int _invoker_id;          // 调用者的ID

int _invoker_moudule_id;  // 调用者的模块id

uint32_t _invoker_ip;      // 调用者的ip

int _call_token;        //调用标识,由调用者设置,返回结果时必须携带此token,否则调用者无法区分是哪一次调用

int _recvier_id;      // 接收者的serverID

uint32_t _req_time    // 请求时间戳

int64_t _random_num;  // 随机数 没有口令配置此项可以不填

uchar _password[32];  // 口令,由 随机数+ 模块id+ 函数名+ 配置文件的口令+ 时间戳 的字符串 一次md5获得,服务器使用同字段md5对比校验,没有口令配置此项留空即可

char _func_name[128]; //函数名

char _text_data[1];  //SttEncoding存储函数体,包括函数名、参数名/参数值

};

```

> _rpc_global_id,_invoker_moudule_id,_invoker_ip这个由调用方直连的第一个MessageServer直接赋值

> _version,_req_time,_random_num,_password,由RPCClient生成,RPCServer校验

> _rpc_retry_count,表示重试的次数,这个由调用方直连的第一个MessageServer发起重试策略时+1

> _rpc_option,包括RPC可达,重试,超时丢弃等标志,现在不可叠加,以后可支持叠加,常见的场景是:

1. RPCClient不太关注返回结果的、关键数据更新类的RPC,建议指定RPC可达。

2. RPCClient非常关注返回结果提示,但又不支持重试的(非幂等RPC),建议指定超时丢弃标志

3. RPCClient非常关注返回结果提示,该RPC又支持幂等,建议指定重试标志

4. _user_data:根据不同的RPC标志,可以指定特定的含义.比如指定最大重试次数或者超时丢弃的时间

> GroupRPCReq(组内随机调用),GroupRPCReq2(组内hash调用)的数据结构也需要同时更新。

3. RPCServer Response基本结构

```

struct RPC_RespNew

{

uint8_t _version;      // 版本号

int64 _rpc_global_id;  // RPC全局唯一id

int recvier_id;        // 接收者的ID;如果是按组接收,此值由MessageServer修改为具体的接收者ID

int invoker_id;        // 调用者的ID

int call_token;        //调用标识,由调用者设置,返回结果时必须携带此token,否则调用者无法区分是哪一次调用

char text_data[1];      //SttEncoding存储调用结果

}

```

#### 以送火箭场景为场景描述架构重构的思路

1. php调用发送火箭RPC接口给鱼翅交易服务器,鱼翅服务器完成RPC调用,并且是把这个消息发送给所有的ChatRoom。

2. 鱼翅交易服务器把发送火箭这个RPC封装成NetMessage通过ChatRoom发送给RoomMaster,RoomMaster找到人气值前50的房间,并向人气值前50的房间的ChatRoom发送火箭广播的NetMessag,ChatRoom再把广播消息发送给MessageRepeater

3. ChatRoom通过NetMessage把发送火箭这个消息事件发送给排行榜服务器

4. ChatRoom通过NetMessage把发送火箭这个消息事件发送给经验服务器

5. ChatRoom发送创建红包RPC给红包服务器

目前的业务流程的主要弊端如下:

1. ChatRoom和大都数服务耦合非常紧密,据我了解,c++的各个小组经常存在同时维护ChatRoom的情况。

2. 同样,新增一个和送火箭相关的服务,ChatRoom也需要增加相关接口。

3. ChatRoom通过RPC、NetMessage和其他业务交互时,如果网络出现抖动或者其他业务在维护或者不稳定时,都会导致数据的丢失,比较影响用户的体验。

针对送火箭这个业务流程,我个人认为比较优雅的架构(新架构)如下:

1. php调用发送火箭RPC接口给鱼翅交易服务器,鱼翅服务器完成RPC调用

2. 鱼翅交易服务器把发送火箭这个RPC封装成消息事件,发送给消息队列服务器。

3. 红包服务器、经验服务器、排行榜服务器、RoomMaster都通过订阅的方式订阅到了发送火箭这个消息。这些服务器按照自己的业务规则处理该消息事件!

4. RoomMaster找到人气值前50的房间,并向人气值前50的房间的ChatRoom发送火箭广播的NetMessag,ChatRoom再把广播消息发送给MessageRepeater。

新架构的优点如下:

1. ChatRoom和其他服务已经完全解耦。

2. 如果新增一个和送火箭相关的服务,鱼翅服务器的逻辑也不用调整。新增的服务只需要订阅送火箭的消息队列就可以了。

3. 消息队列服务器是一个稳定的第三方服务,基本是不用维护的。其他比如红包服务器、经验服务器、排行榜服务器的不稳定,都不会导致数据的丢失。

老框架迁移到新框架下的推进计划:

1. 先挑选送火箭这个业务进行重构,其他业务流程仍然兼容老的RPC的通信方式;

2. 逐步梳理c++组的业务流程,挑选业务流程逐一进行重构;

3. 第一个业务流程的重构预估时间大概在3周左右,后面的每个业务流程重构预估在1周左右,在3-4个月的时间内梳理完所有流程。

## 当前底层框架可以优化的点

1. MessageServer集群可以优雅的增加、删除、修改(同时删除、增加来实现)节点,现在修改某个节点的ip需要重启整个集群?

2. 把弹幕的MessageServer集群和RPC的MessageServer集群分离

3. 协议序列化框架改成ProtoBuffer,可以逐个协议升级

4. MessageServer的通信链路自动检测,防止出现某个节点异常之后很久才发现

## 统一日志

**TODO:本周5和c++组讨论之后再确定**

## 近期之内主要的工作项

WorkItem | 优先级 | 备注

---|---|---

Apr12

【原创】论外部接口调用的设计原则

Author: leeon  Click: 1681   Comments: 0 Category: 架构  Tag: api

      今日后端系统发生一次严重的redis cpu 100%异常问题。期初是怀疑系统硬件问题,因为后端代码已经好几天没有发不过,理应是不会突然故障。于是一步步排查,首先从redis本身入手,其本身redis 占用率过高网上的说法也已经概括的比较全面了,不外乎bgsave,keys模糊查找和滥用排序导致redis性能下降。但是从我们本身的系统架构而言,redis的cpu使用率一直都稳定在3%左右,数据大小本身也不高,rdb文件本身也不大只有几百兆,占用的系统内存页也只有整个硬件内存的20%。磁盘io,cpu都很稳定, 

      首先通过strace定位的方法监控php-fpm和redis进程是否有异样,发现php-fpm都卡在和redis的通信上,而redis进程都卡在频繁读写上。info查看redis的total_commands_processed数值,尽然每秒高达20w次左右,我们知道通过redis的benchmark测试redis的性能通常每秒不走pipleline也差不多这个数值了,业务突然能将redis的吞吐能力消耗干净那肯定不外乎受外部大规模的压力攻击或系统内部的死循环导致。

   攻击肯定就谈不上了,nginx上访问日志很正常,通过redis的monitor命令查看redis的实时执行命令,发现全部都是对同一个key进行写入和查询。接着通过get这个key输出的值发现了端倪。继而定位到业务代码中的执行片断。

     问题排查到现在就好解决了,代码级定位发现了原先设计代码中不合理的设计,对于第三方外部接口调用没有采取异常情况的最终返回,导致不停的递归调用了第三方接口,但是第三方接口又因为查询条件的改变,导致始终执行不成功。这样无法缓存正常数据,就导致不停的查缓存和写缓存操作。

    通过这次问题排查和解决,总结了如何使用第三方接口服务的设计原则:

1. 不管如何设计接口切记尽量避免使用递归来请求接口,如果没有捕获到极端异常,有可能导致程序死循环。

2. 所有外部接口都要设计一个最终失败的逻辑。

3. 对第三方接口增加缓存逻辑的同时,一定要对失败的结果也做一份缓存,避免频繁请求第三方接口而导致同步io延时的问题。

4. 千万不要认为大厂的API接口都是极其稳定的,是的!今天遇到的就是百度的云API提供的接口,虽然大厂通常对自己提供的服务接口都有做API自动化测试,但是故障恢复能力并不一定每个业务团队都能及时响应。更不要以为大厂的开发们都是各个经验丰富,知道如何保证API的可用性和稳定性。如果哪天他们升级了代码将接口的传参做了调整,这种对调用方的伤害也是巨大的。一个设计合理的接口一定遵循一点,同版本接口输入和输出必须保证确定性,做了调整的接口一定得区分版本号。

Mar31

【原创】关于elasticsearch中拼音搜索的性能问题

Author: leeon  Click: 2763   Comments: 0 Category: 架构  Tag: elasticsearch,拼音

近日在elasticsearch按照网上的教程添加拼音支持后发行搜索性能衰减非常大,以前几百万的数据搜索关键字也只需要三四是毫秒,加入了多字段的拼音支持后搜索指定字段性能衰减了6-8倍。类似于网上的配置如下:

在索引分词器中配置如下:

[code="plain"]
index.refresh_interval: 1s
index:
analysis:
tokenizer:
my_pinyin:
type: "pinyin"
first_letter: "none"
padding_char: ""
analyzer:
ik_syno:
type: custom
tokenizer: ik_max_word
filter: [my_synonym_filter]
ik_syno_smart:
type: custom
tokenizer: ik_smart
filter: [my_synonym_filter]
pinyin_analyzer:
tokenizer: my_pinyin
filter: ["word_delimiter","my_ngram"]
py_analyzer:
tokenizer: my_pinyin
filter: ["standard"]
filter:
my_synonym_filter:
type: synonym
synonyms_path: analysis/synonym.txt
ignore_case: true
my_ngram:
type: "nGram"
min_gram: 2
max_gram: 5

[/code]

[code="plain"] {
"folks": {
"properties": {
"name": {
"type": "multi_field",
"fields": {
"name": {
"type": "string",
"store": "no",
"term_vector": "with_positions_offsets",
"analyzer": "pinyin_analyzer",
"boost": 10
},
"primitive": {
"type": "string",
"store": "yes",
"analyzer": "keyword"
}
}
}
}
}
}
[/code]

这种模式配置的字段映射会极大的降低搜索性能,如果想用拼音搜索关键字建议单独设立一个独立的字段来做,不要用multi_field复合字段的方式来配置,这样会大大降低在指定字段中搜索的性能,我猜测如果使用nGram方式来生成分词,会导致生成的token非常多,导致搜索匹配的数据太多导致查询太慢。

分类

标签

归档

最新评论

ligaofeng在11:55:22评论了
file_get_contents超时问题的解决方法
我也不知道叫个啥好在00:59:46评论了
shell中使用while循环ssh的注意事项
司马成空在16:14:13评论了
【原创】ZendStudio中格式化HTML代码错位问题修正
Owen在22:56:46评论了
【原创】MyBatis Generator使用小记
waltye在23:38:05评论了
【原创】武汉互联网公司介绍[2016年8月更新版]

我看过的书

链接

其他

访问本站种子 本站平均热度:6010 c° 本站链接数:27 个 本站标签数:428 个 本站被评论次数:87 次