RPC フレームワークの詳細な例

1、背景

インターネットの発展に伴い、Webサイトアプリケーションの規模は拡大し続けており、従来の垂直アプリケーションアーキテクチャではもはや対応できなくなり、分散サービスアーキテクチャとモバイルコンピューティングアーキテクチャが不可欠となっています。システムはアーキテクチャの秩序ある進化を保証します

単一アプリケーション アーキテクチャ

Web サイトのトラフィックが非常に少ない場合、デプロイ ノードとコストを削減するためにすべての機能を同時にデプロイするのに必要なアプリケーションは 1 つだけです。

このとき、追加、削除、変更、確認の作業負荷を簡略化するデータアクセスフレームワーク(ORM)が鍵

垂直型アプリケーションアーキテクチャ

アクセス数が徐々に増えてきたら、単一のアプリケーションにマシンを追加することによってもたらされる高速化はますます小さくなり、効率を向上させるためにアプリケーションはいくつかの無関係なアプリケーションに分解されます

このとき、フロントエンドページを高速化するためにWebフレームワーク(MVC)が使用されます。開発が鍵です

分散サービスアーキテクチャ

ビジネスラインごとに分割

RPCの悪用をやめ、垂直ビジネス内のローカルjar呼び出しを優先し、ビジネス間でのみRPC呼び出しを使用してください

正確に識別しますビジネス ロジックの所有権を高め、各モジュールの結合力を最大化し、パフォーマンス、可用性、保守性を向上させます。結合を削減します

リリースごとにサーバーの一部のみをデプロイします

各ノードは、さまざまなニーズに応じて拡張および拡張できます

各アプリケーションのアップデート、デプロイメント、操作には影響しません

デプロイメントの分離

チームの分離

データの分離

そしてさらに垂直方向にアプリケーション間の相互作用は避けられず、コア ビジネスは独立したサービスとして抽出され、徐々に安定したサービス センターが形成されるため、フロントエンド アプリケーションは変化する市場の需要により迅速に対応できます

現時点では、分散型ビジネスの再利用と統合を改善するために使用されるサービス フレームワーク (RPC) が鍵です

分散サービス RPC フレームワーク

フロー コンピューティング アーキテクチャ

サービスが増えると、次のような問題が発生します容量評価と小さなサービス リソースの無駄が徐々に現れるため、現時点では、アクセス圧力に基づいてクラスター容量をリアルタイムで管理するディスパッチ センターを追加する必要があります

。マシン使用率を向上させるために使用されるリソース スケジューリングおよびガバナンス センター (SOA) が鍵です

Netty スレッド モデル

Netty のスレッド モデルは、アプリケーションを考慮するため、主に React に基づいています。異なるシナリオにより複数のバージョンが作成されます。

シングルスレッドモード

つまり、サービスリクエストの受信とIO操作の実行がすべて1つのスレッドで完了します。IO多重化などのノンブロッキングIO操作が使用されるため、リクエスト量が少ない場合にシングルスレッドモードが使用されます。いくつかのシーンの問題も解決できます。

単一受信マルチワーカースレッドモード

リクエストの数が増加すると、すべてのIO操作を処理する元の1つのスレッドが対応するパフォーマンス指標をサポートできなくなるため、ワーカースレッドプールの概念が言及されます。この時点では、サービスリクエストを受信したスレッドはまだスレッドです。リクエストを受信した後、リクエストを受信したスレッドは後続のワーカースレッドプールに割り当てられ、スレッドプールからスレッドを取得してユーザーリクエストを実行します。

マルチ受信マルチワーカースレッドモード

リクエスト量がさらに増加すると、サービスリクエストを受信する単一のスレッドではすべてのクライアント接続を処理できなくなるため、サービスリクエストを受信するスレッドもスレッドプールに拡張されます、複数のスレッドで構成されます。スレッドはクライアント接続の受信も担当します。

RPC ビジネス スレッド

上記はすべて Netty 独自のスレッド モデルであり、リクエスト量の増加に伴い継続的に開発されてきた最適化戦略です。 RPC リクエストの場合、アプリケーション システムにとって最も重要なことはビジネス ロジックの処理であり、この種のビジネスは、たとえば、ほとんどのアプリケーションがデータベース操作、Redis、またはその他のネットワーク サービスを伴う場合があります。 、など。ビジネス リクエストにこのような時間のかかる IO 操作がある場合は、ビジネス リクエストを処理するタスクを独立したスレッド プールに割り当てることをお勧めします。そうしないと、netty 自体のスレッドがブロックされる可能性があります。

受信リクエストスレッドと作業スレッドの役割分担

• 受信リクエストスレッドは主にリンクの作成を担当し、その後、リクエストを作業スレッドに委任します

• 作業スレッドは以下を担当しますエンコード、デコード、IO の読み取りおよびその他の操作

ソリューションの実装

現在実装している RPC は、マルチレシーバー マルチワーカー スレッド モードを使用しており、ポートは次のようにサーバー側でバインドされています。 boosGroup はサービス リクエストを受信するために使用されるグループです

workerGroup は特に IO 操作を担当するグループです

ビジネス スレッドを追加するには、ハンドル操作をさらにスレッド プールに委任するだけで済みます。拡張するには、インターフェイスを定義する必要があります。ここ:

スレッド プール インターフェイスを定義する

public void bind(ServiceConfig serviceConfig) {EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(this.rpcServerInitializer)
                    .childOption(ChannelOption.SO_KEEPALIVE,true)
            ;try {ChannelFuture channelFuture = bootstrap.bind(serviceConfig.getHost(),serviceConfig.getPort()).sync();//...channelFuture.channel().closeFuture().sync();


            } catch (InterruptedException e) {throw new RpcException(e);
            }
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

固定サイズのスレッド プールを実装する

Dubbo スレッド プールを参照する

public interface RpcThreadPool {Executor getExecutor(int threadSize,int queues);
}

間奏:

友人が突然、Java の coreSize とは何かと尋ねたことを覚えています。スレッドプールって意味ですか？私は普段マルチスレッドを書かないので、突然ショートしました。私が普段よく使用するデータベーススレッドプールのことを考えると、そのパラメータに非常に感銘を受けましたが、まったく思い出せません。コアサイズ。その後、スレッド プールのいくつかのパラメータを詳しく調べました。この機会にもう一度ショートを避けるために詳しく見てみることができます。

スレッド プール ファクトリ

複数のスレッド プール実装がある場合、スレッド プールはスレッド プール名を通じて動的に選択されます。

@Qualifier("fixedRpcThreadPool")@Componentpublic class FixedRpcThreadPool implements RpcThreadPool {private Executor executor;@Overridepublic Executor getExecutor(int threadSize,int queues) {if(null==executor) {synchronized (this) {if(null==executor) {
                    executor= new ThreadPoolExecutor(threadSize, threadSize, 0L, TimeUnit.MILLISECONDS,
                            queues == 0 ? new SynchronousQueue<Runnable>() :(queues < 0 ? new LinkedBlockingQueue<Runnable>(): new LinkedBlockingQueue<Runnable>(queues)),new RejectedExecutionHandler() {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {                                   //...}
                            });
                }
            }
        }return executor;
    }
}

ChannelHandle の channelRead0 メソッドを変更します

メソッド本体をタスクにラップし、実行のためにスレッド プールに渡します。

@Componentpublic class RpcThreadPoolFactory {@Autowiredprivate Map<String,RpcThreadPool> rpcThreadPoolMap;public RpcThreadPool getThreadPool(String threadPoolName){return this.rpcThreadPoolMap.get(threadPoolName);
    }
}

質問

現在、ストレステストが不足しているため、当面は明確なデータ比較がありません。

以上がRPC フレームワークの詳細な例の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。