我理解的分散式系統
作者簡介
Black，一個喜歡電子和機械的軟體工程師，陰差陽錯的走上了程式猿這條道路。上道之後發現寫程式碼原來那麼有意思，就是頭冷！(￣▽￣)~
前言
說到分散式系統，不得不說集中式系統。傳統集中式系統中整個專案所有的東西都在一個應用裡面。一個網站就是一個應用，當系統壓力較大時，只能橫向擴充套件，增加多個伺服器或者多個容器去做負載均衡，避免單點故障而影響到整個系統。集中式最明顯的優點就是開發，測試，運維會比較方便，不用考慮複雜的分散式環境。弊端也很明顯，系統大而複雜、不易擴充套件、難於維護，每次更新都必須更新所有的應用。
介於集中式系統的種種弊端，促成了分散式系統的形成，分散式系統背後是由一系列的計算機組成，但使用者感知不到背後的邏輯，就像訪問單個計算機一樣。天然的避免了單機故障的問題。應用可以按業務型別拆分成多個應用或服務，再按結構分成介面層、服務層。我們也可以按訪問入口分，如移動端、PC端等定義不同的介面應用。資料庫可以按業務型別拆分成多個例項，還可以對單表進行分庫分表。同時增加分散式快取、訊息佇列、非關係型資料庫、搜尋等中介軟體。分散式系統雖好，但是增加了系統的複雜性，如分散式事務、分散式鎖、分散式session、資料一致性等都是現在分散式系統中需要解決的難題。分散式系統也增加了開發測試運維的成本，工作量增加，其管理不好反而會變成一種負擔。
分散式系統最為核心的要屬分散式服務框架，有了分散式服務框架，我們只需關注各自的業務，而無需去關注那些複雜的服務之間呼叫的過程。
分散式服務框架
目前業界比較流行的分散式服務框架有：阿里的Dubbo、Spring Cloud。這裡不對這些分散式服務框架做對比，簡單的說說他們都做了些什麼，能使我們掉用遠端服務就像掉用本地服務那麼簡單高效。
服務
服務是對使用使用者有功能輸出的模組，以技術框架作為基礎，能實現使用者的需求。比如日誌記錄服務、許可權管理服務、後臺服務、配置服務、快取服務、儲存服務、訊息服務等，這些服務可以靈活的組合在一起，也可以獨立執行。服務需要有介面，與系統進行對接。面向服務的開發，應該是把服務拆分開發，把服務組合執行。更加直接的例子如：歷史詳情、留言板、評論、評級服務等。他們之間能獨立執行，也要能組合在一起作為一個整體。
註冊中心
註冊中心對整個分散式系統起著最為核心的整合作用，支援對等叢集，需要提供 CRUD 介面，支援訂閱釋出機制且可靠性要求非常之高，一般拿zookeeper叢集來做為註冊中心。
分散式環境中服務提供方的服務會在多臺伺服器上部署，每臺伺服器會向註冊中心提供服務方標識、服務列表、地址、對應埠、序列化協議等資訊。註冊中心記錄下服務和服務地址的對映關係，一般一個服務會對應多個地址,這個過程我們稱之為 服務釋出 或 服務註冊 。服務呼叫方會根據服務方標識、服務列表從註冊中心獲取所需服務的資訊（ 地址埠資訊、序列化協議等 ），這些資訊會快取至本地。當服務需要呼叫其它服務時，直接在這裡找到服務的地址，進行呼叫，這個過程我們稱之為 服務發現 。
下面是以zookeeper作為註冊中心的簡單實現：
/**
* 建立node節點
* @param node
* @param data
*/
public boolean createNode(String node, String data) {
try {
byte[] bytes = data.getBytes();
//同步建立臨時順序節點
String path = zk.create(ZkConstant.ZK_RPC_DATA_PATH+"/"+node+"-", bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
log.info("create zookeeper node ({} => {})", path, data);
}catch (KeeperException e) {
log.error("", e);
return false;
}catch (InterruptedException ex){
log.error("", ex);
return false;
}
return true;
}
複製程式碼
如下面zookeeper中寫入的臨時順序節點資訊：
com.black.blackrpc.test.HelloWord （釋出服務時對外的名稱）
00000000010，00000000011 （zk 順序節點id）
127.0.0.1:8888，127.0.0.1:8889 （服務地址埠）
Protostuff （序列化方式） 1.0 （權值，負載均衡策略使用）
這裡使用的是zookeeper的臨時順序節點，為什麼使用臨時順序節點。主要是考慮以下兩點：
一、當服務提供者異常下線時，與zookeeper的連線會中斷，zookeeper伺服器會主動刪除臨時節點，同步給服務消費者。這樣就能避免服務消費者去請求異常的伺服器。
校稿注： 一般消費方也會在實際發起請求前，對當前獲取到的服務提供方節點進行心跳，避免請求連線有問題的節點
二、zk下面是不允許建立2個名稱相同的zk子節點的，通過順序節點就能避免建立相同的名稱。當然也可以不用順序節點的方式,直接以com.black.blackrpc.test.HelloWord建立節點，在該節點下建立資料節點。
下面是zk的資料同步過程：
/**
* 同步節點 （通知模式）
* syncNodes會通過級聯方式，在每次watcher被觸發後，就會再掛上新的watcher。完成了類似鏈式觸發的功能
*/
public boolean syncNodes() {
try {
List<String> nodeList = zk.getChildren(ZkConstant.ZK_RPC_DATA_PATH, new Watcher() {
@Override
public void process(WatchedEvent event) {
if (event.getType() == Event.EventType.NodeChildrenChanged) {
syncNodes();
}
}
});
Map<String,List<String>> map =new HashMap<String,List<String>>();
for (String node : nodeList) {
byte[] bytes = zk.getData(ZkConstant.ZK_RPC_DATA_PATH + "/" + node, false, null);
String key =node.substring(0, node.lastIndexOf(ZkConstant.DELIMITED_MARKER));
String value=new String(bytes);
Object object =map.get(key);
if(object!=null){
((List<String>)object).add(value);
}else {
List<String> dataList = new ArrayList<String>();
dataList.add(value);
map.put(key,dataList);
}
log.info("node: [{}] data: [{}]",node,new String(bytes));
}
/**修改連線的地址快取*/
if(MapUtil.isNotEmpty(map)){
log.debug("invoking service cache updateing....");
InvokingServiceCache.updataInvokingServiceMap(map);
}
return true;
} catch (KeeperException | InterruptedException e) {
log.error(e.toString());
return false;
}
}
複製程式碼
當資料同步到本地時，一般會寫入到本地檔案中，防止因zookeeper叢集異常下線而無法獲取服務提者資訊。
通訊與協議
服務消費者無論是與註冊中心還是與服務提供者，都需要存在網路連線傳輸資料，而這就涉及到通訊。筆者之前也做過這方面的工作，當時使用的是 java BIO 簡單的寫了一個通訊包，使用場景沒有多大的併發,阻塞式的 BIO 也未暴露太多問題。 java BIO 因其建立連線之後會阻塞執行緒等待資料，這種方式必須以一連線一執行緒的方式，即客戶端有連線請求時伺服器端就需要啟動一個執行緒進行處理。當連線數過大時，會建立相當多的執行緒，效能直線下降。
Java NIO: 同步非阻塞，伺服器實現模式為一個請求一個執行緒，即客戶端傳送的連線請求都會註冊到多路複用器上，多路複用器輪詢到連線有I/O請求時才啟動一個執行緒進行處理。
Java AIO: 非同步非阻塞，伺服器實現模式為一個有效請求一個執行緒，客戶端的I/O請求都是由OS先完成了再通知伺服器應用去啟動執行緒進行處理， BIO、NIO、AIO 適用場景分析:
BIO用於連線數目比較小且固定的架構，這種方式對伺服器資源要求比較高，併發侷限於應用中，但程式直觀簡單易理解。
NIO適用於連線數目多且連線比較短（輕操作）的架構，比如聊天伺服器，併發侷限於應用中，程式設計比較複雜，目前主流的通訊框架 Netty、Apache Mina、Grizzl、NIO Framework 都是基於其實現的。
AIO用於連線數目多且連線比較長（重操作）的架構，比如圖片伺服器，檔案傳輸等，充分呼叫OS參與併發操作，程式設計比較複雜。
(有興趣可以看看這篇文章：BIO與NIO、AIO的區別 )
作為基石的通訊，其實要考慮很多東西。如：丟包粘包的情況，心跳機制，斷連重連，訊息快取重發，資源的優雅釋放，長連線還是短連線等。
下面是 Netty 建立服務端，客戶端的簡單實現：
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/***
* netty tcp 服務端
* @author v_wangshiyu
*
*/
public class NettyTcpService {
private static final Logger log = LoggerFactory.getLogger(NettyTcpService.class);
private String host;
private int port;
public NettyTcpService(String address) throws Exception{
String str[] = address.split(":");
this.host=str[0];
this.port=Integer.valueOf(str[1]);
}
public NettyTcpService(String host,int port) throws Exception{
this.host=host;
this.port=port;
}
/**用於分配處理業務執行緒的執行緒組個數 */
private static final int BIZGROUPSIZE = Runtime.getRuntime().availableProcessors()*2; //預設
/** 業務出現執行緒大小*/
private static final int BIZTHREADSIZE = 4;
/*
* NioEventLoopGroup實際上就是個執行緒,
* NioEventLoopGroup在後臺啟動了n個NioEventLoop來處理Channel事件,
* 每一個NioEventLoop負責處理m個Channel,
* NioEventLoopGroup從NioEventLoop數組裡挨個取出NioEventLoop來處理Channel
*/
private static final EventLoopGroup bossGroup = new NioEventLoopGroup(BIZGROUPSIZE);
private static final EventLoopGroup workerGroup = new NioEventLoopGroup(BIZTHREADSIZE);
public void start() throws Exception {
log.info("Netty Tcp Service Run...");
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup);
b.channel(NioServerSocketChannel.class);
b.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
pipeline.addLast("decoder", new ByteArrayDecoder());
pipeline.addLast("encoder", new ByteArrayEncoder());
//pipeline.addLast(new Encoder());
//pipeline.addLast(new Decoder());
pipeline.addLast(new TcpServerHandler());
}
});
b.bind(host, port).sync();
log.info("Netty Tcp Service Success!");
}
/**
* 停止服務並釋放資源
*/
public void shutdown() {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
/**
* 服務端處理器
*/
public class TcpServerHandler extends SimpleChannelInboundHandler<Object>{
private static final Logger log = LoggerFactory.getLogger(TcpServerHandler.class);
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
byte[] data=(byte[])msg;
}
}
複製程式碼
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import io.netty.util.concurrent.Future;
/**
* netty tcp 客戶端
* @author v_wangshiyu
*
*/
public class NettyTcpClient {
private static final Logger log = LoggerFactory.getLogger(NettyTcpClient.class);
private String host;
private int port;
private Bootstrap bootstrap;
private Channel channel;
private EventLoopGroup group;
public NettyTcpClient(String host,int port){
bootstrap=getBootstrap();
channel= getChannel(host,port);
this.host=host;
this.port=port;
}
public String getHost() {
return host;
}
public int getPort() {
return port;
}
/**
* 初始化Bootstrap
* @return
*/
public final Bootstrap getBootstrap(){
group = new NioEventLoopGroup();
Bootstrap b = new Bootstrap();
b.group(group).channel(NioSocketChannel.class);
b.handler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// pipeline.addLast(new Encoder());
// pipeline.addLast(new Decoder());
pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
pipeline.addLast("decoder", new ByteArrayDecoder());
pipeline.addLast("encoder", new ByteArrayEncoder());
pipeline.addLast("handler", new TcpClientHandler());
}
});
b.option(ChannelOption.SO_KEEPALIVE, true);
return b;
}
/**
* 連線，獲取Channel
* @param host
* @param port
* @return
*/
public final Channel getChannel(String host,int port){
Channel channel = null;
try {
channel = bootstrap.connect(host, port).sync().channel();
return channel;
} catch (Exception e) {
log.info(String.format("connect Server(IP[%s],PORT[%s]) fail!", host,port));
return null;
}
}
/**
* 傳送訊息
* @param msg
* @throws Exception
*/
public boolean sendMsg(Object msg) throws Exception {
if(channel!=null){
channel.writeAndFlush(msg).sync();
log.debug("msg flush success");
return true;
}else{
log.debug("msg flush fail,connect is null");
return false;
}
}
/**
* 連線斷開
* 並且釋放資源
* @return
*/
public boolean disconnectConnect(){
//channel.close().awaitUninterruptibly();
Future<?> future =group.shutdownGracefully();//shutdownGracefully釋放所有資源，並且關閉所有當前正在使用的channel
future.syncUninterruptibly();
return true;
}
}
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
/**
* 客戶端處理器
*/
public class TcpClientHandler extends SimpleChannelInboundHandler<Object>{
private static final Logger log = LoggerFactory.getLogger(TcpClientHandler.class);
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
byte[] data=(byte[])msg;
}
}
複製程式碼
說到通訊就不能不說協議，通訊時所遵守的規則，訪問什麼，傳輸的格式等都屬於協議。作為一個開發人員，應該都瞭解TCP/IP協議，它是一個網路通訊模型，以及一整套網路傳輸協議家族，是網際網路的基礎通訊架構。也都應該用過http（超文字傳輸協議），Web伺服器傳輸超文字到本地瀏覽器的傳送協議，該協議建立在TCP/IP協議之上。分散式服務框架服務間的呼叫也會規定協議。為了支援不同場景，分散式服務框架會存在多種協議，如 Dubbo 就支援7種協議： dubbo協議(預設) ， rmi協議 ， hessian協議 ， http協議 ， webservice協議 ， thrift協議 ， memcached協議 ， redis協議 每種協議應對的場景不盡相同，具體場景具體對待。
(這裡詳細介紹了Dubbo 的協議：Dubbo 的7種協議 )
服務路由
分散式服務上線時都是叢集組網部署，叢集中會存在某個服務的多例項，消費者如何從服務列表中選擇合適的服務提供者進行呼叫，這就涉及到服務路由。分散式服務框架需要能夠滿足使用者靈活的路由需求。
透明化路由
很多開源的RPC框架呼叫者需要配置服務提供者的地址資訊，儘管可以通過讀取資料庫的服務地址列表等方式避免硬編碼地址資訊，但是消費者依然要感知服務提供者的地址資訊，這違反了透明化路由原則。而基於服務註冊中心的服務訂閱釋出，消費者通過主動查詢和被動通知的方式獲取服務提供者的地址資訊，而不再需要通過硬編碼方式得到提供者的地址資訊，只需要知道當前系統釋出了那些服務，而不需要知道服務具體存在於什麼位置，這就是透明化路由。
負載均衡
負載均衡策略是服務的重要屬性，分散式服務框架通常會提供多種負載均衡策略，同時支援使用者擴充套件負載均衡策略。
隨機
通常在對等叢集組網中，採用隨機演算法進行負債均衡，隨機路由演算法訊息分發還是比較均勻的，採用JDK提供的 java.util.Random 或者 java.security.SecureRandom 在指定服務提供者列表中生成隨機地址。消費者基於隨機生成的服務提供者地址進行遠端呼叫。
/**
* 隨機
*/
public class RandomStrategy implements ClusterStrategy {
@Override
public RemoteServiceBase select(List<RemoteServiceBase> list) {
int MAX_LEN = list.size();
int index = RandomUtil.nextInt(MAX_LEN);
return list.get(index);
}
}
複製程式碼
隨機還是存在缺點的，可能出現部分節點的碰撞的概率較高，另外硬體配置差異較大時，會導致各節點負載不均勻。為避免這些問題，需要對服務列表加權，效能好的機器接收的請求的概率應該高於一般機器。
/**
* 加權隨機
*/
public class WeightingRandomStrategy implements ClusterStrategy {
@Override
public RemoteServiceBase select(List<RemoteServiceBase> list) {
//存放加權後的服務提供者列表
List<RemoteServiceBase> weightingList = new ArrayList<RemoteServiceBase>();
for (RemoteServiceBase remoteServiceBase : list) {
//擴大10倍
int weight = (int) (remoteServiceBase.getWeight()*10);
for (int i = 0; i < weight; i++) {
weightingList.add(remoteServiceBase);
}
}
int MAX_LEN = weightingList.size();
int index = RandomUtil.nextInt(MAX_LEN);
return weightingList.get(index);
}
}
複製程式碼
輪詢
逐個請求服務地址，到達邊界之後，繼續繞接。主要缺點：慢的提供者會累積請求。例如第二臺機器很慢，但沒掛。當請求第二臺機器時被卡在那。久而久之，所有請求都卡在第二臺機器上。 輪詢策略實現非常簡單，順序迴圈遍歷服務提供者列表，達到邊界之後重新歸零開始，繼續順序迴圈。
/**
* 輪詢
*/
public class PollingStrategy implements ClusterStrategy {
//計數器
private int index = 0;
private Lock lock = new ReentrantLock();
@Override
public RemoteServiceBase select(List<RemoteServiceBase> list) {
RemoteServiceBase service = null;
try {
lock.tryLock(10, TimeUnit.MILLISECONDS);
//若計數大於服務提供者個數,將計數器歸0
if (index >= list.size()) {
index = 0;
}
service = list.get(index);
index++;
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
//兜底,保證程式健壯性,若未取到服務,則直接取第一個
if (service == null) {
service = list.get(0);
}
return service;
}
}
複製程式碼
加權輪詢的話，需要給服務地址新增權重。
/**
* 加權輪詢
*/
public class WeightingPollingStrategy implements ClusterStrategy {
//計數器
private int index = 0;
//計數器鎖
private Lock lock = new ReentrantLock();
@Override
public RemoteServiceBase select(List<RemoteServiceBase> list) {
RemoteServiceBase service = null;
try {
lock.tryLock(10, TimeUnit.MILLISECONDS);
//存放加權後的服務提供者列表
List<RemoteServiceBase> weightingList = new ArrayList<RemoteServiceBase>();
for (RemoteServiceBase remoteServiceBase : list) {
//擴大10倍
int weight = (int) (remoteServiceBase.getWeight()*10);
for (int i = 0; i < weight; i++) {
weightingList.add(remoteServiceBase);
}
}
//若計數大於服務提供者個數,將計數器歸0
if (index >= weightingList.size()) {
index = 0;
}
service = weightingList.get(index);
index++;
return service;
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
//兜底,保證程式健壯性,若未取到服務,則直接取第一個
return list.get(0);
}
}
複製程式碼
服務呼叫時延
消費者快取所有服務提供者的呼叫時延，週期性的計算服務呼叫平均時延。然後計算每個服務提供者服務呼叫時延與平均時延的差值，根據差值大小動態調整權重，保證服務時延大的服務提供者接收更少的訊息，防止訊息堆積。 該策略的特點：保證處理能力強的服務接受更多的訊息，通過動態的權重分配消除服務呼叫時延的震盪範圍，使所有服務的呼叫時延接近平均值，實現負載均衡。
一致性雜湊
相同引數的請求總是傳送到統一服務提供者，當某一臺服務提供者宕機時，原本發往跟提供者的請求，基於虛擬節點，平攤到其他提供者，不會引起劇烈變動，平臺提供預設的虛擬節點數，可以通過配置檔案修改虛擬節點個數。一致性Hash環工作原理如下圖所示：
路由規則
負載均衡只能保證服務提供者壓力的平衡，但是在一些業務場景中需要設定一些過濾規則，比較常用的是基本表示式的條件路由。
通過IP條件表示式配置黑白名單訪問控制：consumerIP != 192.168.1.1。
只暴露部分服務提供者，防止這個叢集服務都被沖垮，導致其他服務也不可用。例如providerIP = 192.168.3*。 讀寫分離：method=find*,list*,get*,query*=>providerIP=192.168.1. 。 前後臺分離：app=web =>providerIP=192.168.1. ,app=java =>providerIP=192.168.2. 。 灰度升級：將WEB前臺應用理由到新的服務版本上：app=web =>provicerIP=192.168.1.*。
由於篇幅原因這裡不細說，還是丟個說的比較詳細的文章地址：服務路由
序列化與反序列化
把物件轉換為位元組序列的過程稱為序列化，把位元組序列恢復為物件的過程稱為反序列化。運程呼叫的時候，我們需要先將Java物件進行序列化，然後通過網路，IO進行傳輸，當到達目的地之後，再進行反序列化獲取到我們想要的結果物件。分散式系統中，傳輸的物件會很多，這就要求序列化速度快，產生位元組序列小的序列化技術。
序列化技術： Serializable, xml, Jackson, MessagePack, fastjson, Protocol Buffer, Thrift,Gson, Avro,Hessian 等
Serializable是java自帶的序列化技術，無法跨平臺，序列化和反序列化的速度相對較慢。
XML技術多平臺支援好，常用於與銀行互動的報文，但是其位元組序列產生較大，不太適合用作分散式通訊框架。
Fastjson是Java語言編寫的高效能的JSON處理器,由阿里巴巴公司開發,位元組序列為json串，可讀性好，序列化也速度非常的快。
Protocol Buffer序列化速度非常快，位元組序列較小，但是可讀性較差。
( 這裡就不一一介紹，有興趣可以看看這篇文章：序列化技術比較 )
一般分散式服務框架會內建多種序列化協議可供選擇，如 Dubbo 支援的7種協議用到的序列化技術就不完全相同。
服務呼叫
本地環境下，使用某個介面很簡單，直接呼叫就行。分散式環境下就不是那麼簡單了，消費者方只會存在介面的定義，沒有具體的實現。想要像本地環境下直接呼叫遠端介面那就得耗費一些功夫了，需要用到 遠端代理 。
下面是我盜的圖：
通訊時序如下：
消費者端沒有具體的實現，需要呼叫介面時會動態的去建立一個代理類。與spirng整合的情況，那直接在bean構建的時候注入代理類。
下面是構建代理類：
import java.lang.reflect.Proxy;
public class JdkProxy {
public static Object getInstance(Class<?> cls){
JdkMethodProxy invocationHandler = new JdkMethodProxy();
Object newProxyInstance = Proxy.newProxyInstance(
cls.getClassLoader(),
new Class[] { cls },
invocationHandler);
return (Object)newProxyInstance;
}
}
複製程式碼
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
public class JdkMethodProxy implements InvocationHandler {
@Override
public Object invoke(Object proxy, Method method, Object[] parameters)throws Throwable {
//如果傳進來是一個已實現的具體類
if (Object.class.equals(method.getDeclaringClass())) {
try {
return method.invoke(this, parameters);
} catch (Throwable t) {
t.printStackTrace();
}
//如果傳進來的是一個介面
} else {
//實現介面的核心方法
//return RemoteInvoking.invoking(serviceName, serializationType, //timeOut,loadBalanceStrategy,method, parameters);
}
return null;
}
}
複製程式碼
代理會做很多事情，對請求服務的名稱及引數資訊的的序列化、通過路由選擇最為合適服務提供者、建立通訊連線傳送請求資訊（或者直接發起http請求）、最後返回獲取到的結果。當然這裡面需要考慮很多問題，如呼叫超時，請求異常，通訊連線的快取，同步服務呼叫還是非同步服務呼叫等等。
同步服務呼叫：客戶端發起遠端服務呼叫請求，使用者執行緒完成訊息序列化之後，將訊息投遞到通訊框架，然後同步阻塞，等待通訊執行緒傳送請求並接收到應答之後，喚醒同步等待的使用者執行緒，使用者執行緒獲取到應答之後返回。
非同步服務呼叫：基於JAVA的 Future 機制，客戶端發起遠端服務呼叫請求，該請求會被標上 requestId ,同時建立一個與 requestId 對應 Future ，客戶端通過 Future 的 get方法獲取結果時會被阻塞。服務端收到請求應達會回傳 requestId ，通過 requestId 去解除對應 Future 的阻塞，同時set對應結果，最後客戶端獲取到結果。
構建Future，以requestId為key，put到執行緒安全的map中。get結果時需要寫入timeOut超時時間，防止由於結果的未返回而導致的長時間的阻塞。
SyncFuture<RpcResponse> syncFuture =new SyncFuture<RpcResponse>();
SyncFutureCatch.syncFutureMap.put(rpcRequest.getRequestId(), syncFuture);
try {
RpcResponse rpcResponse= syncFuture.get(timeOut,TimeUnit.MILLISECONDS);return rpcResponse.getResult();
}catch (Exception e){
throw e;
}finally {
SyncFutureCatch.syncFutureMap.remove(rpcRequest.getRequestId());
}
複製程式碼
結果返回時通過回傳的requestId獲取對應Future寫入Response,Future執行緒解除阻塞。
log.debug("Tcp Client receive head："+headAnalysis+"Tcp Client receive data：" +rpcResponse);
SyncFuture<RpcResponse> syncFuture= SyncFutureCatch.syncFutureMap.get(rpcResponse.getRequestId());
if(syncFuture!=null){
syncFuture.setResponse(rpcResponse);
}
複製程式碼
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
public class SyncFuture<T> implements Future<T> {
// 因為請求和響應是一一對應的，因此初始化CountDownLatch值為1。
private CountDownLatch latch = new CountDownLatch(1);
// 需要響應執行緒設定的響應結果
private T response;
// Futrue的請求時間，用於計算Future是否超時
private long beginTime = System.currentTimeMillis();
public SyncFuture() {
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return false;
}
@Override
public boolean isCancelled() {
return false;
}
@Override
public boolean isDone() {
if (response != null) {
return true;
}
return false;
}
// 獲取響應結果，直到有結果才返回。
@Override
public T get() throws InterruptedException {
latch.await();
return this.response;
}
// 獲取響應結果，直到有結果或者超過指定時間就返回。
@Override
public T get(long timeOut, TimeUnit unit) throws InterruptedException {
if (latch.await(timeOut, unit)) {
return this.response;
}
return null;
}
// 用於設定響應結果，並且做countDown操作，通知請求執行緒
public void setResponse(T response) {
this.response = response;
latch.countDown();
}
public long getBeginTime() {
return beginTime;
}
}
複製程式碼
SyncFuture<RpcResponse> syncFuture =new SyncFuture<RpcResponse>(); SyncFutureCatch.syncFutureMap.put(rpcRequest.getRequestId(), syncFuture); RpcResponse rpcResponse= syncFuture.get(timeOut,TimeUnit.MILLISECONDS); SyncFutureCatch.syncFutureMap.remove(rpcRequest.getRequestId()); 複製程式碼
除了同步服務呼叫，非同步服務呼叫，還有並行服務呼叫，泛化呼叫等呼叫形式
( 這裡就不做介紹，有興趣可以看看這篇文章： 服務框架多形式的服務呼叫：同步、非同步、並用、泛化 )
高可用
簡單的介紹了下分散式服務框架，下面來說下分散式系統的高可用。一個系統設計開發出來，三天兩晚就出個大問題，導致無法使用，那這個系統也不是什麼好系統。業界流傳一句話："我們系統支援X個9的可靠性"。這個X是代表一個數字，X個9表示在系統1年時間的使用過程中，系統可以正常使用時間與總時間（1年）之比。
3個9：(1-99.9%)*365*24=8.76小時，表示該系統在連續執行1年時間裡最多可能的業務中斷時間是8.76小時，4個9即52.6分鐘，5個9即5.26分鐘。要做到如此高的可靠性，是非常大的挑戰。一個大型分散式專案可能是由幾十上百個專案構成，涉及到的服務成千上萬，主鏈上的一個流程就需要流轉多個團隊維護的專案。拿4個9的可靠性來說，平攤到每個團隊的時間可能不到10分鐘。這10分鐘內需要頂住壓力，以最快的時間找到並解決問題，恢復系統的可用。
下面說說為了提高系統的可靠性都有哪些方案：
服務檢測：某臺伺服器與註冊中心的連線中斷，其提供的服務也無響應時，系統應該能主動去重啟該服務，使其能正常對外提供。
故障隔離：叢集環境下，某臺伺服器能對外提供服務，但是因為其他原因，請求結果始終異常。這時就需要主動將該節點從叢集環境中剔除，避免繼續對後面的請求造成影響，非高峰時期再嘗試修復該問題。至於機房故障的情況，只能去遮蔽整個機房了。目前餓了麼做的是異地多活，即便單邊機房掛了，流量也可以全量切換至另外一邊機房，保證系統的可用。
監控：包含業務監控、服務異常監控、db中介軟體效能的監控等，系統出現異常的時候能及時的通知到開發人員。等到線下報上來的時候，可能影響已經很大了。
壓測：產線主鏈路的壓測是必不可少的，單靠整合測試，有些高併發的場景是無法覆蓋到的，壓測能暴露平常情況無法出現的問題，也能直觀的提現系統的吞吐能力。當業務激增時，可以考慮直接做系統擴容。
sop方案與演練：產線上隨時都可能會發生問題，抱著出現問題時再想辦法解決的態度是肯定不行的，時間根本來不及。提前做好對應問題的sop方案，能節省大量時間，儘快的恢復系統的正常。當然平常的演練也是不可少的，一旦產線故障可以做到從容不迫的去應對和處理。
除了上述方案外，還可以考慮 服務策略 的使用：
降級策略
業務高峰期，為了保證核心服務，需要停掉一些不太重要的業務，如雙十一期間不允許發起退款(*￣▽￣)、只允許檢視3個月之內的歷史訂單等業務的降級，呼叫服務介面時，直接返回的空結果或異常等服務的降級,都屬於分散式系統的降級策略。服務降級是可逆操作，當系統壓力恢復到一定值不需要降級服務時，需要去除降級，將服務狀態恢復正常。 服務降級主要包括 遮蔽降級 和 容錯降級 ：
遮蔽降級:分散式服務框架直接遮蔽對遠端介面的請求，不發起對遠端服務的呼叫，直接返回空結果、丟擲指定異常、執行本地模擬介面實現等方式。
容錯降級：非核心服務不可呼叫時，可以對故障服務做業務放通，保證主流程不受影響。如請求超時、訊息解碼異常、系統擁塞保護異常， 服務提供方系統異常等情況。 筆者之前就碰到過因雙方沒有做容錯降級導致的系統故障的情況。午高峰時期，對方呼叫我們的一個非核心查詢介面，我們系統因為bug問題一直異常，導致對方呼叫這個介面的頁面異常而無法跳轉到主流程頁面，影響了產線的生產。當時對方緊急發版才使系統恢復正常。
限流策略
說到限流，最先想到的就是秒殺活動了，一場秒殺活動的流量可能是正常流量的幾百至幾千倍，如此高的流量系統根本無法處理，只能通過限流來避免系統的崩潰。服務的限流本質和秒殺活動的限流是一樣的，都是限制請求的流入，防止服務提供方因大量的請求而崩潰。
限流演算法： 令牌桶、漏桶、計數器演算法
上述演算法適合單機的限流，但涉及到整個叢集的限流時，得考慮使用快取中介軟體了。例如：某個服務1分鐘內只允許請求2次，或者一天只允許使用1000次。由於負載均衡存在，可能叢集內每臺機器都會收到請求，這種時候就需要快取來記錄呼叫方某段時間內的請求次數，再做限流處理。redis就很適合做此事。限流演算法的實現
熔斷策略
熔斷本質上是一種過載保護機制，這一概念來源於電子工程中的斷路器，當電流過大時，保險絲會熔斷，從而保護整個電路。同樣在分散式系統中，當被呼叫的遠端服務無法使用時，如果沒有過載保護，就會導致請求的資源阻塞在遠端伺服器上耗盡資源。很多時候，剛開始可能只是出現了局部小規模的故障，然而由於種種原因，故障影響範圍越來越大，最終導致全域性性的後果。當下遊服務因訪問壓力過大而響應變慢或失敗，上游服務為了保護自己以及系統整體的可用性，可以暫時切斷對下游服務的呼叫。
熔斷器的設計思路
Closed：初始狀態，熔斷器關閉，正常提供服務
Open: 失敗次數，失敗百分比達到一定的閾值之後，熔斷器開啟，停止訪問服務
Half-Open：熔斷一定時間之後，小流量嘗試呼叫服務，如果成功則恢復，熔斷器變為 Closed 狀態
資料一致性
一個系統設計開發出來，必須保證其執行的資料準確和一致性。拿支付系統來說：使用者銀行卡已經扣款成功，系統裡卻顯示失敗，沒有給使用者的虛擬帳戶充值上，這會引起客訴。說的再嚴重點，使用者發起提現，資金已經轉到其銀行賬戶，系統卻沒扣除對應虛擬帳號的餘額，直接導致資金損失了。如果這時候使用者一直髮起提現，那就酸爽了。
CAP原則
說到資料一致性，就不得不說到 CAP原則 。 CAP原則 中指出任何一個分散式系統中， Consistency（一致性 C） 、 Availability（可用性 A） 、 Partition tolerance（分割槽容錯性P） ，三者不可兼得。傳統單機資料庫基於 ACID特性（原子性（Atomicity）、一致性（Consistency）、隔離性（Isolation）、永續性（Durability）） ，放棄了分割槽容錯性，能做到可用性和一致性。對於一個分散式系統而言，分割槽容錯性是一個最基本的要求。既然是一個分散式系統，那麼分散式系統中的元件必然需要被部署到不同的節點，會出現節點與節點之間的網路通訊，而網路問題又是一定會出現的異常情況， 分割槽容錯性 也就成為了一個分散式系統必然需要面對和解決的問題。系統架構師往往需要把精力花在如何根據業務特點在 一致性 和 可用性 之間尋求平衡。
集中式系統，通過資料庫事務的控制，能做到資料的強一致性。但是分散式系統中，涉及多服務間的呼叫，通過分散式事務的方案： 兩階段提交（2PC） 、 三階段提交（3PC） 、**補償事務（TCC）**等雖然能實現資料的強一致，但是都是通過犧牲可用性來實現。
BASE理論
BASE理論是對 CAP原則 中一致性和可用性權衡的結果： Basically Available（基本可用）、Soft state（軟狀態）和Eventually consistent（最終一致性） 。BASE理論，其來源於對大規模網際網路系統分散式實踐的總結，是基於 CAP原則 逐步演化而來的。其最核心思想是：即使無法做到強一致性，但每個應用都可以根據自身業務特點，採用適當的方式來使系統達到最終一致性。
基本可用
基本可用是指分散式系統在出現不可預知故障的時候，允許損失部分可用性，這不等價於系統不可用。
軟狀態
軟狀態指允許系統中的資料存在中間狀態，並認為該中間狀態的存在不會影響系統的整體可用性，即允許系統在不同節點的資料副本之間進行資料同步的過程存在延時
最終一致性
最終一致性強調的是所有的資料副本，在經過一段時間的同步之後，最終都能夠達到一致的狀態。因此，最終一致性的本質是需要系統保證最終資料能夠達到一致，而不需要實時保證系統資料的強一致性。
總的來說， BASE理論 面向的是大型高可用可擴充套件的分散式系統，和傳統的事物 ACID特性 是相反的，它完全不同於 ACID 的強一致性模型，而是通過犧牲強一致性來獲得可用性，並允許資料在一段時間內是不一致的，但最終達到一致狀態。同時，在實際的分散式場景中，不同業務單元和元件對資料一致性的要求是不同的，因此在具體的分散式系統架構設計過程中，ACID特性和BASE理論往往又會結合在一起。
結尾
分散式系統涉及到的東西還有很多，如：分散式鎖、定時排程、資料分片、效能問題、各種中介軟體的使用等。筆者分享只是瞭解到的那一小部分的知識而已。之前本著學習的目的也寫過一個非常簡單的分散式服務框架 blackRpc ，通過它瞭解了分散式服務框架內部的一些活動。本文中所有程式碼都能在該專案中找到，有興趣讀者可以看看。
閱讀部落格還不過癮？
