RPC Proxy

开发环境

基于Thrift的RPC Proxy

RPC的分层

L1层(应用层)

# Transport层是所有的RPC服务都共用的(除非在同一个请求内部实现了并发)
_ = get_base_protocol(settings.RPC_LOCAL_PROXY)
protocol = get_service_protocol("typo")
from cy_typo.services.TypoService import Client
_typo_client = Client(protocol)
# 函数调用
rpc_result = _typo_client.correct_typo(content)
content = rpc_result.fixed
# 或者使用Pool
pool = ConnectionPool(10, "127.0.0.1:5550")
with pool.base_protocol() as base_protocol:
 protocol = get_service_protocol("typo", base_protocol)
 client = Client(protocol)

L2层(Proxy层)

# Go中的deamon似乎不太容易实现,借助: nohup &可以实现类似的效果(Codis也如此)
# 默认的proxy(绑定: 127.0.0.0:5550)
nohup rpc_proxy -c config.ini -L proxy.log >/dev/null 2>&1 &
# 在测试服务器上部署,给大家开发测试使用的proxy
nohup rpc_proxy -c config_test.ini -L proxy_test.log >/dev/null 2>&1 &

L3层(负载均衡)

## 配置文件
config.ini
zk=rd1:2181,rd2:2181,mysql2:2181
# 线上的product一律以: online开头
# 而测试的product禁止使用 online的服务
product=test
verbose=0
zk_session_timeout=30
## Load Balance
service=typo
front_host=
front_port=5555
back_address=127.0.0.1:5556
# 使用网络的IP, 如果没有指定front_host, 则使用使用当前机器的内网的Ip来注册
ip_prefix=10.
## Server
worker_pool_size=2
## Client/Proxy
proxy_address=127.0.0.1:5550
# Go中的deamon似乎不太容易实现,借助: nohup &可以实现类似的效果(Codis也如此)
nohup rpc_lb -c config.ini -L lb.log >/dev/null 2>&1 &

L3层也可以直接就是Rpc服务层,例如:

func main() {
	proxy.RpcMain(BINARY_NAME, SERVICE_DESC,
		// 默认的ThriftServer的配置checker
		proxy.ConfigCheckThriftService,
		// 可以根据配置config来创建processor
		func(config *utils.Config) proxy.Server {
			// 可以根据配置config来创建processor
			alg.InitGeoFix(FILE_GPS_2_SOGOU)
			processor := gs.NewGeoLocationServiceProcessor(&alg.GeoLocationService{})
			return proxy.NewThriftRpcServer(config, processor)
		})
}

L4层

class TypoProcessor(object):
 def correct_typo(self, query):
 # print "Query: ", query
 result = TypoResult()
 new_query, mappings = typo_manager.correct_typo(query)
 result.fixed = new_query
 result.fixes = []
 for old, new in mappings:
 result.fixes.append(FixedTerm(old, new))
 return result
config_path = "config.ini"
config = parse_config(config_path)
endpoint = config["back_address"]
service = config["service"]
worker_pool_size = int(config["worker_pool_size"])
processor = Processor(TypoProcessor())
s = RpcWorker(processor, endpoint, pool_size=worker_pool_size, service=service)
s.connect(endpoint)
s.run()

系统架构

运维部署