Netty là một framework giao tiếp bất đồng bộ hiệu năng cao, cung cấp một API đơn giản hơn so với các giao diện NIO gốc của Java. Tuy nhiên, không phải tất cả các nhà phát triển đều hiểu rõ cơ chế hoạt động bên trong của Netty khi sử dụng.

Việc tìm kiếm các bài viết hướng dẫn xây dựng connection pool phía client với Netty cũng khá hạn chế. Bài viết này sẽ giới thiệu cách xây dựng connection pool hiệu quả bằng Netty.

Trước tiên, hãy xem xét ví dụ client mẫu của Netty:

// Tạo một EventLoopGroup, có thể coi là một thread pool trong Netty, mặc định số lượng thread tối đa là 2 lần số lượng bộ xử lý.
EventLoopGroup group = new NioEventLoopGroup();
try {
    // Lớp hỗ trợ thiết lập kết nối của Netty.
    Bootstrap b = new Bootstrap();
    // Cấu hình các thuộc tính và thêm handler vào pipeline.
    b.group(group)
     .channel(NioSocketChannel.class)
     .option(ChannelOption.TCP_NODELAY, true)
     .handler(new ChannelInitializer<SocketChannel>() {
         @Override
         public void initChannel(SocketChannel ch) throws Exception {
             ChannelPipeline p = ch.pipeline();
             if (sslCtx != null) {
                 p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
             }
             // p.addLast(new LoggingHandler(LogLevel.INFO));
             p.addLast(new EchoClientHandler());
         }
     });

    // Khởi tạo kết nối.
    ChannelFuture f = b.connect(HOST, PORT).sync();

    // Chặn cho đến khi kết nối bị đóng.
    f.channel().closeFuture().sync();
} finally {
    group.shutdownGracefully();
}

Ví dụ trên khá đơn giản. Nhưng nếu bạn cần kết nối tới 100 máy chủ, bạn sẽ làm thế nào? Cách tiếp cận sau đây có thể xuất hiện trong suy nghĩ của bạn:

// Duyệt qua danh sách các Host cần kết nối.
for(Host host : hosts){
      // Tạo một EventLoopGroup mới cho mỗi kết nối.
      EventLoopGroup group = new NioEventLoopGroup(1); // Chỉ một thread cho mỗi EventLoopGroup
      try {
          Bootstrap b = new Bootstrap();
          b.group(group)
           .channel(NioSocketChannel.class)
           .option(ChannelOption.TCP_NODELAY, true)
           .handler(new ChannelInitializer<SocketChannel>() {
               @Override
               public void initChannel(SocketChannel ch) throws Exception {
                   ChannelPipeline p = ch.pipeline();
                   if (sslCtx != null) {
                       p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
                   }
                   p.addLast(new EchoClientHandler());
               }
           });

      ChannelFuture f = b.connect(HOST, PORT).sync();
      f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

Vấn đề rõ ràng ở đây là: nếu mỗi kênh (channel) tương ứng với một NioEventLoopGroup riêng, chúng ta sẽ tạo ra mô hình connection:thread = 1:1. Khi số lượng kết nối tăng lên, vấn đề "thread inflation" (phình to luồng) sẽ trở nên nghiêm trọng.

Giải quyết vấn đề "Thread Inflation"

Để tránh vấn đề này, chúng ta chỉ cần sửa đổi mã nguồn một chút:

// Chỉ tạo một EventLoopGroup dùng chung cho tất cả các kết nối.
NioEventLoopGroup group = new NioEventLoopGroup();
Bootstrap b = new Bootstrap();

try {
  b.group(group)
   .channel(NioSocketChannel.class)
   .option(ChannelOption.TCP_NODELAY, true)
   .handler(new ChannelInitializer<SocketChannel>() {
       @Override
       public void initChannel(SocketChannel ch) throws Exception {
           ChannelPipeline p = ch.pipeline();
           if (sslCtx != null) {
               p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
           }
           p.addLast(new EchoClientHandler());
       }
     });

 // Sử dụng cùng một Bootstrap để tạo nhiều kết nối.
 for(Host HOST : Hosts){
     ChannelFuture f = b.connect(HOST, PORT).sync();
 }
 // Các hoạt động tiếp theo với các kết nối đã được thiết lập...
} finally {
    // Đóng EventLoopGroup khi không còn sử dụng.
    // group.shutdownGracefully(); // Có thể cần điều chỉnh logic đóng gói tùy theo yêu cầu.
}

Trong đoạn mã trên, chúng ta sử dụng cùng một đối tượng Bootstrap để thiết lập nhiều kết nối, cho phép chúng chia sẻ chung một NioEventLoopGroup. Điều này giúp tránh được vấn đề phình to luồng.

Các vấn đề tiếp theo và giải pháp

Tuy nhiên, vấn đề chưa hoàn toàn được giải quyết. Chúng ta cần xem xét các yêu cầu sau:

1. Làm thế nào để mỗi kết nối sử dụng một Handler khác nhau nếu cần?
2. Làm thế nào để tái sử dụng các kết nối, tránh việc tạo mới Channel liên tục?

Để xây dựng một connection pool cho các hoạt động bất đồng bộ, chúng ta cần một mô hình khác. Netty cung cấp FixedChannelPool để hỗ trợ điều này.

Dưới đây là mã giả minh họa cách sử dụng FixedChannelPool:

Bootstrap bootstrap = new Bootstrap()
                        .channel(NioSocketChannel.class)
                        .group(new NioEventLoopGroup()); // Sử dụng một EventLoopGroup chung.

// Khởi tạo ChannelPoolHandler tùy chỉnh.
ChannelPoolHandler customPoolHandler = new CustomChannelPoolHandler(); // Cần định nghĩa lớp này.

// Tạo một FixedChannelPool với số lượng kết nối cố định.
FixedChannelPool pool = new FixedChannelPool(bootstrap, customPoolHandler, MAX_CONNECTIONS); // MAX_CONNECTIONS là số kết nối tối đa.

// Lấy một kết nối từ pool hoặc tạo mới nếu cần.
ChannelFuture acquireFuture = pool.acquire();

// Thêm listener để xử lý khi kết nối sẵn sàng.
acquireFuture.addListener(new GenericFutureListener<Future<Channel>>() {
    @Override
    public void operationComplete(Future<Channel> future) throws Exception {
        if (future.isSuccess()) {
            Channel channel = future.getNow();
            // Thực hiện các thao tác trên channel đã lấy được.
            // ...

            // Trả lại kết nối vào pool khi hoàn thành.
            // pool.release(channel);
        } else {
            // Xử lý lỗi khi không lấy được kết nối.
        }
    }
});

Kết luận

Trong hầu hết các trường hợp, việc sử dụng connection pool có sẵn (ví dụ: HikariCP cho JDBC) là đủ. Tuy nhiên, đối với một số kịch bản kinh doanh đặc thù (như thời gian phản hồi không xác định), việc xây dựng một connection pool bất đồng bộ bằng Netty là cần thiết. Lựa chọn giải pháp phù hợp với bối cảnh nghiệp vụ là yếu tố then chốt.