Первый взгляд на нетти

Как клиент-серверная среда NIO, netty может быстро и легко создавать сетевые приложения, такие как серверы протоколов и клиенты. Netty использует опыт реализации FTP, SMTP, HTTP и других протоколов и обеспечивает надежность и ремонтопригодность программы на основе простоты использования и гибкости.

Когда мы впервые начали изучать сетевое программирование на Java, мы все открылиsocketпорт, а затем вызовите методaccept()Метод блокирует ожидание подключения, а затем продолжает считывать данные. После того, как мы освоим больше позже, мы попытаемся прочитать данные неблокирующим способом и использоватьSelectorНеблокирующий ввод-вывод для селекторов.

Поскольку бизнес продолжает расти, тысячи одновременных томов больше не являются невозможными. Надежная и простая в использовании среда разработки на стороне клиента стала целью разработчиков для обеспечения более высокой пропускной способности и масштабируемой производительности, и netty идеально отвечает потребностям людей. Он инкапсулирует сложный базовый API Java и предоставляет его простым в использовании способом.Использование netty позволяет уделить больше внимания разработке бизнес-логики, а не тривиальной базовой архитектуре.

Ниже приведены основные компоненты netty, подробности будут записаны позже:

Channel(ряд).

каналjava nioОсновная концепция , представляющая рабочее соединение с объектом (например, сетевое соединение, файловые операции ввода-вывода).
Перезвони.

Обратный вызов на самом деле является методом, и при вызове метода также будет вызываться его указанный (или связанный) метод обратного вызова.
Future.

Futureявляется заполнителем для асинхронной операции, когда асинхронная операция завершается, ее соответствующийFutureобъект будет называться. неттиFutureвыполнить--ChannelFutureПозволяет асинхронной операции регистрировать несколькоChannelFutureListenerпример. Каждый исходящий ввод-вывод netty возвращаетChannelFutureпример.
событие иHandler(процессор).

Netty использует события, чтобы уведомлять нас об изменениях в рабочем состоянии.События могут включать:
- Активация или деактивация соединения (входящее событие).
- Чтение данных (входящее событие).
- Пользовательские события (входящие события).
- События ошибок (входящие события).
- Открытие или закрытие соединения с удаленным узлом (исходящее событие).
- Записать данные в сокет (исходящее событие).
неттиChannelHandlerЕсть много реализаций, и вы также можете настроить реализацию. Каждое событие отправляется в соответствующийChannelHandlerметод в классе.

Примечание. Входящие и исходящие относительны.ChannelHandlerДругими словами, введитеChannelHandlerдля входящих, отChannelHandlerОтправка сообщения исходящая.

пример кода

Вот простое использование netty:

Реализация входящего обработчика

@ChannelHandler.Sharable // 标识一个 channelHandler 可以被多个 channel 安全地调用。
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    // 当有入站消息时该方法就会调用
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buffer = (ByteBuf)msg;
        System.out.println("服务器收到消息:" + buffer.toString(CharsetUtil.UTF_8));
        // 将接收到的消息写给发送者，而不冲刷出站消息。
        ctx.write(buffer);
    }

    @Override
    // channelRead消费完读取的数据的时候被触发
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 将未决消息冲刷到远程节点，并且关闭该 channel
        ChannelFuture channelFuture = ctx.writeAndFlush(Unpooled.EMPTY_BUFFER);
        channelFuture.addListener(ChannelFutureListener.CLOSE);
    }

    @Override
    // 在读操作时处理异常
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 打印异常栈，并关闭该 channel
        cause.printStackTrace();
        ctx.close();
    }
}

Примечание. Обработка обработчика сервера netty осуществляется в виде цепочки ответственности. По умолчанию обработчик каналов перенаправляет вызовы методов следующему обработчику каналов в цепочке. еслиexceptionCaught()метод не реализован где-то в цепочке, то полученное исключение передается вchannelPipelineзаканчиваются и записываются.

реализация сервера

public class NettyServer {
    public static void main(String[] args) {
        NettyServer nettyServer = new NettyServer();
        nettyServer.start(8888);
    }

    public void start(int port){
        // 处理TCP连接请求
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 处理I/O事件
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            // 用于引导和绑定服务器
            ServerBootstrap bootstrap = new ServerBootstrap();
            //将上面的线程组加入到 bootstrap 中
            bootstrap.group(bossGroup,workGroup)
                    //将通道设置为异步的通道
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // 因为 NettyServerHandler 被标注为 @Sharable，所以可以使用相同的实例
                            socketChannel.pipeline().addLast(new NettyServerHandler());
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG,200)
                    .childOption(ChannelOption.SO_KEEPALIVE,true);
            // 异步绑定服务器，调用 sync() 方法阻塞等待直到绑定完成。
            ChannelFuture future = bootstrap.bind(port).sync();
            // 获取 channel 的 closeFuture，并且阻塞直到它完成。
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

Реализация исходящего обработчика

@ChannelHandler.Sharable
public class NettyClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
       ctx.writeAndFlush(Unpooled.copiedBuffer("netty 活跃", CharsetUtil.UTF_8));
    }

    // 记录已接收的消息存储
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        System.out.println("客户端接收消息：" + msg.toString(CharsetUtil.UTF_8));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

SimpleChannelInboundHandlerиChannelInboundHandler:

Использование клиентаSimpleChannelInboundHandlerПричина в том, что ему не нужно учитывать асинхронные операции, когдаchannelRead0После выполнения методаSimpleChannelInboundHandlerотпустит указатель, чтобы сохранить сообщениеByteBufferпамяти.

использование сервераChannelInboundHandlerпотому что ему нужно доставлять сообщения клиенту, иctx.write()метод асинхронный, возможноchannelRead()После выполнения метода он не вернулся, поэтому во избежание такой ситуации используйтеChannelInboundHandler.channelReadCompleteметод будет вchannelRead()Запускается, когда прочитанные данные потребляются, и в этот момент он сбрасывает вывод вchannel.

Реализация клиента

public class NettyClient {
    public static void main(String[] args) {
        NettyClient nettyClient = new NettyClient();
        nettyClient.connect("localhost", 8888);
    }

    public void connect(String hostname,int port) {
        // 处理TCP连接请求
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            // 用于引导和绑定服务器
            Bootstrap bootstrap = new Bootstrap();
            //将上面的线程组加入到 bootstrap 中
            bootstrap.group(group)
                	//将通道设置为异步的通道
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline()
                                    .addLast(new NettyClientHandler());
                        }
                    })
                    .option(ChannelOption.TCP_NODELAY,true);

            // 连接到远程节点，阻塞等待直到连接完成。
            ChannelFuture future = bootstrap.connect(hostname, port).sync();
            // 阻塞，直到 channel 关闭。
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 关闭线程池并释放所有资源。
            group.shutdownGracefully();
        }
    }
}

Суммировать

Из приведенного выше примера кода мы можем видеть, что конструкция сервера netty и клиента на самом деле аналогична, оба из которых реализуют процессор, а затем связывают его.

Разработка сервера:

Создание и реализация логики процессора.
1. Приложение масштабируется по запросуChannelHandler.
2. Вызывается на разные мероприятияChannelHandler.
СоздайтеServerBootstrapinstance для начальной загрузки и привязки сервера.
создать и назначитьNioEventLoopGroupэкземпляр для обработки запросов.
создать и назначитьNioEventLoopGroupэкземпляр для обработки событий.
Указывает локальную привязку к серверуInetSocketAddress.
Инициализировать каждый новый с помощью экземпляра обработчикаchannel.
перечислитьServerBootstrap.bind()способ привязки к серверу.

Развитие клиента:

Создание и реализация логики процессора.
1. Приложение масштабируется по запросуChannelHandler.
2. Вызывается на разные мероприятияChannelHandler.
Создайте экземпляр Bootstrap для инициализации клиента.
создать и назначитьNioEventLoopGroupэкземпляр для обработки событий.
Создать для сервераInetSocketAddressпример.
Когда соединение установлено,Handlerбудет установлен вchannelизChannelPiplineначальство.
перечислитьBootstrap.connect()способ подключения к удаленному узлу.

Впереди у нас есть предварительное представление о netty, включая его основное содержание и простое использование. Далее мы продолжим изучать основные компоненты и концепции дизайна netty.

чистый дизайн компонентов

Интерфейс канала

Мы знаем, что основные компоненты netty включают в себя каналы Каналы — очень важная концепция в java, которая выражает связь с операциями сущностей. Netty абстрагирует несколько операций как каналы, в том числе:

Работа сокета.
Многопоточность.
Асинхронное уведомление.

Мы знаем, что основные операции ввода-вывода в сети (установление соединения, чтение данных, запись данных) зависят от интерфейса, предоставляемого базовой сетевой передачей, которая представлена в виде класса Socket в java. Netty дополнительно инкапсулирует этот класс, что значительно снижает сложность класса Socket и предоставляет ряд реализаций, основанных на интерфейсе Channel:

EmbeddedChannel
EpollDatagramChannel
LocalServerChannel
KQueueDatagramChannel
NioDatagramChannel
NioSctpChannel
NioSocketChannel

Как показано на рисунке, каждому каналу будут назначены ChannelPipline и ChannelConfig.ChannelConfig содержит всю конфигурацию канала и поддерживает горячее обновление. Обычно при создании экземпляра Channel создается ChannelConfig по умолчанию:

// NioServerSocketChannel构造方法
public NioServerSocketChannel(ServerSocketChannel channel) {
    super(null, channel, SelectionKey.OP_ACCEPT);
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

Netty предоставляет класс ChannelOption, определяющий все типы параметров, поддерживаемые ChannelConfig, которые можно использовать следующим образом:

NioServerSocketChannel channel = new NioServerSocketChannel();
ServerSocketChannelConfig config = channel.config();
config.setOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
// 针对每个ChannelOption选项，netty还提供了对应的方法，比如上面的设置可以这样代替
// config.setAllocator(PooledByteBufAllocator.DEFAULT);
// 设置通道
bootstrap.channel(channel.getClass());

Чтобы обеспечить порядок канала, он реализует интерфейс Comparable. Таким образом, если два разных канала имеют одинаковое значение хеш-функции, будет выдана ошибка.

Канал также предоставляет другие методы, основные из них:

имя метода	описывать
eventLoop	Возвращает соответствующий EventLoop
pipeline	Возвращает соответствующий ChannelPipline
localAddress	Возвращает локальный SocketAddress
remoteAddress	Возвращает удаленный SocketAddress
write	Запишите данные на удаленный узел, эти данные будут переданы в ChannelPipline и поставлены в очередь до тех пор, пока они не будут сброшены.
flush	Сбросить ранее записанные данные на удаленный узел
writeFlush	Запишите данные и сбросьте их на удаленный узел
isActive	Определить, активен ли канал

Транспорты, поддерживаемые netty

NIO

NIO реализует все асинхронные операции ввода-вывода через Selector, который работает в потоке, который проверяет изменения состояния и отвечает соответствующим образом. Битовый шаблон селектора определяется java.nio.channels.SelectionKey:

название	описывать
OP_ACCEPT	Получайте уведомления, когда соединение принято и канал создан
OP_CONNECT	Получайте уведомления, когда соединение установлено
OP_READ	Получайте уведомления, когда данные могут быть прочитаны
OP_WRITE	Получать уведомления, когда буфер отправки канала доступен для записи

Epoll

Epoll — это высокопроизводительная масштабируемая функция уведомления о событиях ввода-вывода от Linux. Netty предоставляет соответствующий API для Linux epoll.

OIO

OIO (old IO) — это старый блокирующий ввод-вывод, который используется через обычный транспортный API и может использоваться для перехода при переносе проекта.

Local

Это собственный транспорт, предоставляемый netty для асинхронной связи между клиентами и серверами, работающими в одной и той же JVM.

Embedded

Встроенная передача позволяет нам встраивать набор ChannelHandler в качестве вспомогательных классов внутри других ChannelHandler, чтобы функциональность ChannelHandler можно было расширить без изменения внутреннего кода. Ключом к встроенной передаче является реализация канала EmbeddedChannel.

Жизненный цикл канала

Канал определяет набор шаблонов состояний, связанных с ChannelInboundHandler:

государство	описывать
ChannelUnregistered	Канал создан, но не зарегистрирован в EventLoop
ChannelRegistered	Канал зарегистрирован в EventLoop
ChannelActive	Канал активен и может принимать и отправлять данные
ChannelInactive	Канал неактивен

Модульное тестирование на основе EmbeddedChannel

EmbeddedChannel предоставляется netty специально для улучшения модульного тестирования ChannelHandler. Его можно использовать для имитации отправки и запроса сообщений для проверки функциональной реализации соответствующего ChannelHandler.

EmbeddedChannel предоставляет следующие часто используемые API:

API	описывать
writeInbound	Записывает входящее сообщение в EmbeddedChannel. Возвращает true, если данные могут быть прочитаны из EmbeddedChannel через readInbound();
readInbound	Чтение входящих сообщений от EmbeddedChannel. Любой возврат проходит через весь ChannelPipeline. Если чтение не готово, этот метод возвращает null;
writeOutbound	Напишите исходящее сообщение в EmbeddedChannel. Возвращает true, если данные могут быть прочитаны из EmbeddedChannel через readOutbound();
readOutbound	Чтение исходящего сообщения от EmbeddedChannel. Любой возврат проходит через весь ChannelPipeline. Если чтение не готово, этот метод возвращает null;
Finish	Если данные могут быть прочитаны из входящего или исходящего трафика, пометьте EmbeddedChannel как завершенный и вернитесь. Это также вызывает метод закрытия EmbeddedChannel;

@Test
public void testFramesDecoded(){
    ByteBuf buf= Unpooled.buffer();
    for (int i=0;i<9;i++){
        buf.writeByte(i);
    }
    ByteBuf input=buf.duplicate();
    EmbeddedChannel channel=new EmbeddedChannel(
        new FixedLengthFrameDecoder(3)
    );
    assertTrue(channel.writeInbound(input.retain()));
    assertTrue(channel.finish());    
    //读取消息
    ByteBuf read=channel.readInbound();
    assertEquals(buf.readSlice(3),read);
    read.release();

    read = (ByteBuf) channel.readInbound();
    assertEquals(buf.readSlice(3), read);
    read.release();
    read = (ByteBuf) channel.readInbound();
    assertEquals(buf.readSlice(3), read);
    read.release();
    assertNull(channel.readInbound());
    buf.release();
}

Интерфейс EventLoop

EventLoop определяет основную абстракцию netty для обработки того, что происходит в течение жизненного цикла соединения.

Из приведенного выше рисунка мы видим, что:

Группа EventLoopGroup содержит один или несколько циклов EventLoop.
EventLoopGroup может быть привязан только к одному потоку в своем жизненном цикле.
Все события ввода-вывода, обрабатываемые EventLoop, будут обрабатываться в его выделенном потоке.
Канал может быть зарегистрирован только в одном EventLoop в течение своего времени существования.
EventLoop может быть назначен одному или нескольким каналам.

netty использует цикл событий EventLoop для обработки задач запроса в соединении. EventLoop использует два основных API: сеть и параллельное программирование. Пакет io.netty.util.concurrent основан на пакете JUC для предоставления исполнителей потоков. Классы пакета io.util.channel расширяют эти классы и интерфейсы для взаимодействия с событиями канала.

Планирование задач Netty расширяет ScheduledExecutorService JUC, потому что запланированные задачи netty могут быть помещены в очередь выполнения EventLoop, и нет необходимости выполнять переключение потоков, как JUC, поэтому потребление производительности снижается:
ctx.channel().eventLoop().schedule(new Runnable() { // 创建任务处理线程
            @Override
            public void run() {
                System.out.println("EventLoop任务调度");
            }
        }, 60, TimeUnit.MICROSECONDS); // 指定调度周期

Интерфейс ChannelFuture

Netty предоставляет интерфейс ChannelFuture в качестве заполнителя для асинхронных вызовов.Метод ChannelFuture.addListener() регистрирует ChannelFutureListener, который будет уведомлен о завершении операции.

Интерфейс ChannelHandler

Интерфейс ChannelHandler становится контейнером для логики приложения, обрабатывающего входящие и исходящие данные.

Netty предоставляет ряд реализаций ChannelHandler по умолчанию в режиме адаптера, предназначенных для упрощения разработки логики обработки приложений. Ниже приведены классы адаптеров, которые часто используются при написании пользовательских обработчиков каналов:

ChannelHandlerAdapter
ChannelInbouAdapter
ChannelOutboundHandlerAdapter
ChannelDuplexHandler

У ChannelHandler есть соответствующий жизненный цикл. Когда ChannelHandler добавляется или удаляется из ChannelPipline, будут вызываться соответствующие операции.

тип	описывать
handlerAdded	ChannelHandler добавлен в ChannelPipline
handlerRemoved	ChannelHandler удален ChannelPipline
exceptionCaught	Произошла ошибка в ChannelPipline во время обработки

Мы можем настроить собственную логику обработки, реализуя интерфейсы ChannelInboundHandler или ChannelOutboundHandler или расширяя классы ChannelInboundHandlerAdapter или ChannelOutboundHandlerAdapter.

Управление ресурсами: Когда процессор обрабатывает данные, нам нужно убедиться, что в конце нет утечки ресурсов.Упомянутая позже технология подсчета ссылок ByteBuf также призвана решить эту проблему. Netty предоставляет ResourceLeakDetector для обнаружения утечек памяти. Уровень утечки может быть пройден черезjava -Dio.netty.leakDetectionLevel=泄漏级别указать. Уровни утечки, определенные netty:

уровень	описывать
DISABLED	Отключить обнаружение утечек
SIMPLE	Используйте частоту дискретизации по умолчанию 1% и сообщайте об обнаруженных утечках (уровень по умолчанию).
ADVANCED	Используйте частоту дискретизации 1 % и сообщайте об обнаруженных утечках и местах доступа к сообщениям.
PARANOID	Выборка каждого сообщения, отчет об обнаруженной утечке и соответствующем месте доступа

Кодировщики и декодеры являются типичными реализациями ChannelHandler.

При разработке серверов мы должны обратить внимание на решение проблемы с липкими пакетами. Общее решение состоит в том, чтобы определить формат протокола и проанализировать данные в соответствии с протоколом после получения информации. Netty соответствует разным форматам и предоставляет разные типы абстракций.XxxDecoderиXxxEncoder, такие как ProtobufEncoder и ProtobufDecoder, которые поддерживают протокольные буферы Google.

Кодировщики декодирования Netty можно условно разделить на две категории:

байттомессаже
От сообщения к сообщению (MessageToMessage)

Мы можем реализовать собственный процессор, расширив декодеры и кодировщики, предварительно созданные netty. Для каждого сообщения, прочитанного из входящего канала, после завершения метода channelRead() будет вызываться метод decode(), предоставленный декодером, и пересылать декодированные байты следующему обработчику ChannelInboundHandler в ChannelPipline. То же самое касается исходящих.

декодер

netty предоставляет базовый класс для реализации кодировщика байт-сообщение: ByteToMessageDecoder (наследует ChannelInboundHandlerAdapter ), который буферизует входящие данные до тех пор, пока они не будут готовы к обработке. Он имеет два наиболее важных метода:

API	описывать
decode(ChannelHandlerContext ctx,ByteBuf in,List< Object > out)	Абстрактный метод, который необходимо реализовать. Метод decode() вызывается с ByteBuf, содержащим входящие данные, и списком, в который добавляется декодированное сообщение. Вызовы этого метода будут повторяться до тех пор, пока не будет определено, что в список не были добавлены новые элементы или что в ByteBuf больше нет доступных для чтения байтов. Затем, если список не пуст, его содержимое будет передано следующему обработчику ChannelInboundHandler в ChannelPipeline.
decodeLast(ChannelHandlerContext ctx,ByteBuf in,List< Objec t> out)	Этот метод будет вызываться один раз, когда состояние канала становится неактивным. По умолчанию вызывается метод decode().

API

описывать

decode(ChannelHandlerContext ctx,ByteBuf in,List< Object > out)

Абстрактный метод, который необходимо реализовать. Метод decode() вызывается с ByteBuf, содержащим входящие данные, и списком, в который добавляется декодированное сообщение. Вызовы этого метода будут повторяться до тех пор, пока не будет определено, что в список не были добавлены новые элементы или что в ByteBuf больше нет доступных для чтения байтов. Затем, если список не пуст, его содержимое будет передано следующему обработчику ChannelInboundHandler в ChannelPipeline.

decodeLast(ChannelHandlerContext ctx,ByteBuf in,List< Objec t> out)

Этот метод будет вызываться один раз, когда состояние канала становится неактивным. По умолчанию вызывается метод decode().

ReplayingDecoder расширяет ByteToMessageDecoder (наследует ChannelInboundHandlerAdapter), ReplayingDecoder не нужно оценивать длину полученных данных при обработке данных. ReplayingDecoder реализует собственный ReplayingDecoderByteBuf, выдает исключение, когда данных недостаточно, затем ReplayingDecoder сбрасывает readerIndex и снова вызывает метод декодирования.

Хотя ReplayingDecoder более удобен в использовании, чем ByteToMessageDecoder, на самом деле ReplayingDecoder работает немного медленнее, чем ByteToMessageDecoder.

netty предоставляет базовый класс для реализации кодировщика сообщений: MessageToMessageDecoder. Он имеет наиболее важные методы:

API	описывать
decode(ChannelHandlerContext ctx,ByteBuf in,List< Object > out)	Абстрактный метод, который необходимо реализовать. Метод decode() вызывается с ByteBuf, содержащим входящие данные, и списком, в который добавляется декодированное сообщение. Вызовы этого метода будут повторяться до тех пор, пока не будет определено, что в список не были добавлены новые элементы или что в ByteBuf больше нет доступных для чтения байтов. Затем, если список не пуст, его содержимое будет передано следующему обработчику ChannelInboundHandler в ChannelPipeline.

API

описывать

decode(ChannelHandlerContext ctx,ByteBuf in,List< Object > out)

Netty предоставляет исключение TooLongFrameException, которое выдается, когда декодер превышает указанный предел размера, не позволяя декодеру буферизировать большой объем данных и вызывая нехватку памяти.

Декодирование протоколов на основе разделителей: протокол сообщений с разделителями использует определенные символы для обозначения начала или конца сообщения или сегмента сообщения (кадра). Декодеры для обработки протоколов на основе разделителей и на основе длины:

название	описывать
DelimiterBasedFrameDecoder	Общий декодер для извлечения кадров с использованием любого пользовательского разделителя.
LineBasedFrameDecoder	Декодер, извлекающий кадры, разделенные символами конца строки (\n или \r\n). Этот декодер работает быстрее, чем DelimiterBasedFrameDecoder.

Протоколы декодирования на основе длины: протоколы, основанные на длине, определяют кадр, кодируя его длину в заголовке кадра, а не используя специальный разделитель для обозначения его конца. Декодеры для протоколов на основе длины:

название	описывать
FixedLengthFrameDecoder	Извлечь кадр фиксированной длины, указанный при вызове конструктора
LengthFieldBasedFrameDecoder	Извлечь кадр на основе значения длины, закодированного в заголовке кадра; смещение и длина этого поля задаются в конструкторе

Кодер

Кодер реализует ChannelOutboundHandler и преобразует исходящие данные из одного формата в другой.

MessageToByteEncoder — это базовый класс для преобразования сообщений в байты, наиболее важными методами являются:

API	описывать
encode(ChannelHandlerContext ctx,I msg,ByteBuf out)	Абстрактный метод, который необходимо реализовать. Вызывается с исходящим сообщением (типа I), которое должно быть закодировано этим классом как ByteBuf. Затем ByteBuf будет перенаправлен следующему ChannelOutboundHandler в ChannelPipeline.

Причина, по которой ByteToMessageDecoder имеет больше методов decodeLast, чем MessageToByteEncoder, заключается в том, что декодер обычно должен генерировать последнее сообщение после закрытия канала.

MessageToMessageEncoder — это базовый класс для преобразования сообщений в сообщения, наиболее важными методами являются:

API	описывать
encode(ChannelHandlerContext ctx,I msg,List< Object > out)	Абстрактный метод, который необходимо реализовать. . Каждое сообщение, написанное с помощью метода write(), будет передано методу encode() для кодирования как одно или несколько исходящих сообщений. Затем эти исходящие сообщения будут переадресованы следующему обработчику ChannelOutboundHandler в ChannelPipeline.

API

описывать

encode(ChannelHandlerContext ctx,I msg,List< Object > out)

Абстрактный метод, который необходимо реализовать. . Каждое сообщение, написанное с помощью метода write(), будет передано методу encode() для кодирования как одно или несколько исходящих сообщений. Затем эти исходящие сообщения будут переадресованы следующему обработчику ChannelOutboundHandler в ChannelPipeline.

кодек

Из вышеизложенного мы знаем, что кодировщик наследует ChannelInboundHandlerAdapter, а декодер наследует ChannelOutboundHandlerAdapter.Если мы реализуем эти две функции одновременно, можем ли мы интегрировать кодирование и декодирование в один класс? Netty в основном предоставляет нам кодеки для байтов и сообщений.

Кодек Byte: абстрактный класс ByteToMessageCodec, который объединяет ByteToMessageDecoder и MessageToByteEncode, этот класс состоит из трех важных методов:

API	описывать
decode(ChannelHandlerContext ctx,ByteBuf in,List< Object > out)	Этот метод будет вызываться всякий раз, когда есть байты, доступные для потребления. Он преобразует входящий ByteBuf в указанный формат сообщения и перенаправляет его следующему обработчику ChannelInboundHandler в ChannelPipeline.
decodeLast(ChannelHandlerContext ctx,ByteBuf in,List< Object > out)	Реализация этого метода по умолчанию делегирует методу decode(). Он будет вызываться только один раз, когда состояние канала становится неактивным. Его можно переопределить для реализации специальной обработки.
encode(ChannelHandlerContext ctx,msg,ByteBuf out)	Этот метод будет вызываться для каждого сообщения (типа I), которое будет закодировано и записано в исходящий ByteBuf.

Кодек сообщения: абстрактный класс MessageToMessageCodec, с помощью MessageToMessageCodec процесс двустороннего обмена этого преобразования может быть реализован в одном классе. Два важных метода этого класса:

API	описывать
protected abstract decode(ChannelHandlerContext ctx,INBOUND_IN msg,List< Object > out)	Этот метод вызывается с сообщением типа INBOUND_IN. Он декодирует их в сообщения типа OUTBOUND_IN, которые будут перенаправлены следующему обработчику ChannelInboundHandler в ChannelPipeline.
protected abstract encode(ChannelHandlerContext ctx,OUTBOUND_IN msg,List< Object > out)	Этот метод будет вызываться для каждого сообщения типа OUTBOUND_IN. Эти сообщения будут закодированы как сообщения типа INBOUND_IN, а затем перенаправлены следующему обработчику ChannelOutboundHandler в ChannelPipeline.

Класс CombinedChannelDuplexHandler может комбинировать кодировщики и декодеры, как показано в следующем примере.

// ByteToCharDecoder是自定义的编码器，CharToByteEncoder是自定义的解码器
public class CombinedByteCharCodec extends
    CombinedChannelDuplexHandler<ByteToCharDecoder, CharToByteEncoder> {
    public CombinedByteCharCodec() {
    super(new ByteToCharDecoder(), new CharToByteEncoder());
    }
}

Сериализация и десериализация

Netty предоставляет три метода сериализации:

JDK поставляется с сериализацией.

название	описывать
CompatibleObjectDecode	Декодер для взаимодействия с удаленными узлами, отличными от Netty, с использованием сериализации JDK.
CompatibleObjectEncoder	Кодер для взаимодействия с удаленными узлами, отличными от Netty, с использованием сериализации JDK.
ObjectDecoder	Декодер, созданный на основе сериализации JDK, который использует пользовательскую сериализацию для декодирования; он обеспечивает повышение скорости при отсутствии других внешних зависимостей. В противном случае предпочтительны другие реализации сериализации.
ObjectEncoder	Кодер, созданный на основе сериализации JDK для кодирования с использованием пользовательской сериализации; он обеспечивает повышение скорости при отсутствии других внешних зависимостей. В противном случае предпочтительны другие реализации сериализации.

Сериализация с использованием JBoss: JBoss не только устраняет некоторые проблемы с сериализатором, поставляемым с jdk, но и повышает производительность.

название	описывать
CompatibleMarshallingDecoder,CompatibleMarshallingEncoder	Совместимость с удаленными узлами, использующими только сериализацию JDK.
MarshallingDecoder,MarshallingEncoder	Применяется к узлам, использующим JBoss Marshalling. Эти классы должны использоваться вместе

Сериализация с использованием протокольных буферов. Протокольные буферы — это формат обмена данными, разработанный и открытый компанией Google.

название	описывать
ProtobufDecoder	Расшифровать сообщение с помощью protobuf
ProtobufEncoder	Кодировать сообщения с помощью protobuf
ProtobufVarint32FrameDecoder	Динамически разделить полученный ByteBuf в соответствии с целочисленным значением поля Google Protocol Buffers «Base 128 Varints» в сообщении.
ProtobufVarint32LengthFieldPrepender	Добавляет целочисленное значение поля длины буфера протокола Google «Base 128 Varints» в начало ByteBuf.

Предустановленный обработчик каналов

использовать https: netty использует пакет javax.net.ssl через SslHandler для реализации SSL-шифрования. Кроме того, он также предоставляет реализацию SSLEngine инструментов OpenSSL — OPenSSLEngine, которая имеет лучшую производительность, чем SSLEngine JDK. Netty попытается загрузить OpenSSLEngine по умолчанию, а если это не удастся, загрузит JdkSSLEngine. Связанные с ним методы:

API	описывать
setHandshakeTimeout (long,TimeUnit),setHandshakeTimeoutMillis (long),getHandshakeTimeoutMillis()	Установите и получите тайм-аут.По истечении тайм-аута рукопожатие ChannelFuture будет уведомлено о сбое
setCloseNotifyTimeout (long,TimeUnit),setCloseNotifyTimeoutMillis (long),getCloseNotifyTimeoutMillis()	Установите и получите таймаут, по истечении которого сработает уведомление о закрытии и соединение будет закрыто. Это также приведет к сбою уведомления ChannelFuture.
handshakeFuture()	Возвращает ChannelFuure, который будет уведомлен, когда рукопожатие будет завершено. Если рукопожатие было выполнено ранее, возвращается ChannelFuture, содержащий результат предыдущего рукопожатия.
close(),close(ChannelPromise),close(ChannelHandlerContext,ChannelPromise)	Отправьте close_notify, чтобы запросить закрытие и уничтожение базового SslEngine.

HTTP-кодек:netty предоставляет несколько обработчиков каналов для форматирования данных в HTTP-ответы или преобразования запросов в данные. На следующем рисунке показаны компоненты ответа на http-запрос:

Основные кодеки:

название	описывать
HttpRequestEncoder	Кодировать сообщения HttpRequest, HttpContent и LastHttpContent в байты
HttpResponseEncoder	Кодировать сообщения HttpResponse, HttpContent и LastHttpContent в байты
HttpRequestDecoder	Декодировать байты в сообщения HttpRequest, HttpContent и LastHttpContent.
HttpResponseDecoder	Декодировать байты в сообщения HttpResponse, HttpContent и LastHttpContent.

http-агрегация: Для некоторых данных запроса или ответа кодек netty может быть не в состоянии разобрать их полностью, а разобрать их на несколько фрагментов данных.Например, HttpServerCodec может получить параметры только в uri, то если использовать почтовый запрос, т.к. информация сохраняется в messageBody, поэтому ее нельзя полностью проанализировать. В это время вам нужно добавить HttpObjectAggregator. Структура HttpObjectAggregator выглядит следующим образом:

В функции decode() MessageAggregator есть параметр currentMessage, который является переменной-членом обработчика. Каждый канал соответствует экземпляру обработчика. В этом currentMessage будут храниться результаты нескольких итераций декодирования, что является ключом к реализации агрегации. .

HTTP-сжатие: несмотря на то, что сжатие данных по протоколу HTTP увеличивает нагрузку на часы сервера, оно может сэкономить сетевой трафик и увеличить скорость передачи. Клиент может использовать HttpContentDecompressor для обработки содержимого с сервера, а сервер может использовать HttpContentCompressor() для сжатия данных.

websocket: netty предоставляет множество фреймворков для реализации длинных соединений через веб-сокеты. Ниже приведены типы websocketFrame:

название	описывать
BinaryWebSocketFrame	кадр данных: двоичные данные
TextWebSocketFrame	кадр данных: текстовые данные
ContinuationWebSocketFrame	dataframe: текстовые или двоичные данные, принадлежащие предыдущему BinaryWebSocketFrame или TextWebSocketFrame
CloseWebSocketFrame	Кадр управления: запрос CLOSE, код состояния закрытия и причина закрытия.
PingWebSocketFrame	Кадр управления: запрос PongWebSocketFrame
PongWebSocketFrame	Кадр управления: ответ на запрос PingWebSocketFrame

Чтобы повысить безопасность WebSocket, просто добавьте SslHandler в качестве первого ChannelHandler для в ChannelPipeline.

управление соединением:netty предоставляет менеджер для обнаружения незанятых соединений и соединений с тайм-аутом. Основные из них:

название	описывать
IdleStateHandler	IdleStateEvent запускается, когда соединение бездействует слишком долго. Затем событие IdleStateEvent можно обработать, переопределив метод userEventTriggered() в ChannelInboundHandler.
ReadTimeoutHandler	Если в течение указанного интервала времени входящие данные не получены, создается исключение ReadTimeoutException и соответствующий канал закрывается. Исключение ReadTimeoutException можно обнаружить, переопределив метод exceptionCaught() в ChannelHandler.
WriteTimeoutHandler	Если исходящие данные не записываются в течение указанного интервала времени, создается исключение WriteTimeoutException и соответствующий канал закрывается. Вы можете обнаружить WriteTimeoutException, переопределив метод exceptionCaught() вашего ChannelHandler.

Интерфейс ChannelPipline

Интерфейс ChannelPipline реализует связанные вызовы и может добавлять контейнеры ChannelHandler. При создании канала он автоматически назначается каналу ChannelPipline, которому он принадлежит.

Как показано на рисунке, при поступлении входящего сообщения оно начинается с заголовка ChannelPipline и передается первому ChannelInboundHandler.Когда этот ChannelHandler обрабатывается, он передается следующему ChannelInboundHandler, пока не достигнет ChannelPipline. Исходящее сообщение аналогично входящему.

Когда ChannelHandler добавляется в ChannelPipline, ему назначается ChannelHandlerContext, который представляет привязку между ChannelHandler и ChannelPipline.

В netty есть два способа отправки сообщений:

Пишите напрямую в канал, что приведет к потоку сообщений из хвоста ChannelPipline.

Напишите непосредственно в ChannelHandlerContext, который запустит поток сообщений из следующего ChannelHandlerContext в ChannelPipline.

Когда ChannelPipline распространяет событие, он проверяет, соответствует ли тип следующего ChannelHandler в ChannelPipline направлению движения события, и если не совпадает, то ChannelHandler пропускает.

ChannelHandlerContext

Основная функция ChannelHandlerContext состоит в том, чтобы заставить ChannelHandler взаимодействовать с ChannelPipline.ChannelHandler может уведомлять следующий ChannelHandler о ChannelPipline, которому принадлежит ChannelHandler, и может изменять ChannelPipline, которому он принадлежит.

API-интерфейс ChannelHandlerContext:

В ChannelHandlerContext Channel и ChannelPipline есть несколько методов, следует отметить, что если эти методы вызываются в Channel или ChannelPipeline, они будут распространяться по всему ChannelPipeline. пока вызов находится в ChannelHandlerContext тот же метод на , начнется с текущего связанного ChannelHandler и будет распространяться только на Следующий ChannelHandler в ChannelPipeline, который может обработать событие.

ServerBootstrap/Boostrap

Класс начальной загрузки Netty предоставляет контейнер для конфигурации сетевого уровня приложения, ServerBootstrap используется для начальной загрузки сервера, а Bootstrap используется для начальной загрузки клиента.

Мы видим, что и ServerBootstrap, и Bootstrap реализуют абстрактный класс AbstractBootstrap.

В предыдущей статье при реализации начальной загрузки сервера мы создали и связали две группы EventLoopGroups.Почему? Глядя на исходный код, мы можем увидеть его объяснение:

Интерпретация исходного кода заключается в том, что при загрузке сервера требуется EventLoopGroup, чтобы указать, что сам сервер был привязан к сокету, на котором прослушивается локальный порт, а вторая группа используется для обработки клиентских подключений.

Глядя на исходный код Bootstrap, мы видим, что при вызове Bootstrap.group(EventLoopGroup group) фактически вызывается групповой метод AbstractBootstrap, который совпадает с первой строкой ServerBootstrap.group(EventLoopGroup parentGroup, EventLoopGroup childGroup ).

ByetBuf

Базовой единицей сетевой передачи являются байты, и для операций с байтами можно использовать Java ByetBuffer API. Но у ByteBuf есть некоторые ограничения:

Его длина фиксирована.После завершения выделения емкость не может динамически расширяться и сокращаться.Когда кодируемый объект POJO превышает емкость байтового буфера, возникает исключение индекса за пределами границ;
ByteBuffer имеет только позицию указателя, которая идентифицирует элемент управления позицией.При чтении и записи вам нужно вручную вызывать flip() и rewind() и т. д., что может легко привести к сбою обработки программы.
Функции API ограничены, и некоторые расширенные и практичные функции должны быть реализованы самими пользователями.

Для этого netty снова строится поверх ByteBuffer, обеспечивая новый и мощный API ByteBuf, который имеет следующие преимущества:

Может быть расширен пользовательскими типами буферов.
Нулевое копирование достигается за счет встроенных составных буферов.
емкость растет.
Чтение и запись используют разные указатели.
Цепочка методов поддерживается.
Поддерживается подсчет ссылок.
Объединение поддерживается.

Шаблон использования ByteBuf

буфер кучи

Режим буфера кучи также известен как режим резервного массива. Данные хранятся в куче JVM, что достигается за счет хранения данных в массиве. Пример выглядит следующим образом:

public static void heapBuffer() {
    // 创建Java堆缓冲区
    ByteBuf heapBuf = Unpooled.buffer(); 
    if (heapBuf.hasArray()) { // 判断是否有支撑数组
        byte[] array = heapBuf.array(); // 获取该支撑数组引用
        int offset = heapBuf.arrayOffset() + heapBuf.readerIndex(); // 计算第一个字节偏移量
        int length = heapBuf.readableBytes(); // 获取可读字节数
        handleArray(array, offset, length); // 调用自己的方法
    }
}

прямой буфер

Прямой буфер принадлежит прямой памяти, выделенной вне кучи, и не занимает емкость кучи. Он подходит для процесса передачи сокета, избегая процесса копирования данных из внутреннего буфера в прямой буфер, и имеет лучшую производительность. Его главный недостаток заключается в том, что они более дороги в выделении и свободны, чем буферы на основе кучи, а поскольку данные находятся не в куче java, перед обработкой необходимо сделать еще одну копию.

public static void directBuffer() {
    ByteBuf directBuf = Unpooled.directBuffer();
    if (!directBuf.hasArray()) { // 如果不是堆缓冲区
        int length = directBuf.readableBytes(); // 获取可读字节数
        byte[] array = new byte[length]; // 分配一个新数组来保存数据
        directBuf.getBytes(directBuf.readerIndex(), array); // 将数据复制到新数组
        handleArray(array, 0, length); // 调用自己的方法
    }
}

составной буфер

Составные буферы — это буферы, специфичные для сети. По сути, аналогично предоставлению комбинированного представления одного или нескольких ByteBuf, различные типы ByteBuf могут быть добавлены и удалены по мере необходимости. Составной буфер не поддерживает доступ к своему резервному массиву. Поэтому, если вы хотите получить доступ, вам нужно скопировать содержимое в динамическую память перед доступом к нему.

public static void byteBufComposite() {
    // 复合缓冲区，只是提供一个视图
    CompositeByteBuf messageBuf = Unpooled.compositeBuffer();
    ByteBuf headerBuf = Unpooled.buffer();
    ByteBuf bodyBuf = Unpooled.directBuffer();
    messageBuf.addComponents(headerBuf, bodyBuf); // 将ByteBuf实例追加到CompositeByteBuf
    messageBuf.removeComponent(0); // 移除索引位置为0的缓冲区
    for (ByteBuf buf : messageBuf) { // 遍历缓冲区
        System.out.println(buf.toString());
    }
}

операции на уровне байтов

индекс произвольного доступа: индекс ByteBuf начинается с 0.

public static void byteBufRelativeAccess() {
    ByteBuf buffer = Unpooled.buffer();
    for (int i = 0; i < buffer.capacity(); i++) {
        byte b = buffer.getByte(i);// 不改变readerIndex值
        System.out.println((char) b);
    }
}

Для тех методов, которым требуется только параметр значения индекса, они не изменяют readIndex и writeIndex, но их можно изменить вручную, вызвав readerIndex(index) и writeIndex(index).

отбрасываемые байты: Область отбрасываемых байтов относится к области между [0, readerIndex). Вызовите метод discardReadBytes(), чтобы отбросить уже прочитанные байты. Метод discardReadBytes() перемещает содержимое области читаемых байтов (CONTENT). Если он вызывается часто, возникнут многочисленные накладные расходы на репликацию данных, что окажет определенное влияние на производительность.

читаемые байты: Доступная для чтения область байтов относится к области между [readerIndex, WriterIndex). любойreadxxx()иskipxxx()Метод операции изменит индекс readerIndex.

доступные для записи байты: Доступная для записи область байтов относится к области между [writerIndex, емкость). любойwritexxx()Методы действия изменят значение WriterIndex.

Управление индексами:

markReaderIndex(), markWriterIndex() - отметить текущую позицию потока.
resetReaderIndex(), resetWriterIndex() - сбрасывает поток в отмеченную позицию.
readIndex(index), writeIndex(int) - переместить индекс чтения/записи в указанную позицию.
clear() - устанавливает readerIndex и WriterIndex в 0, но содержимое памяти не очищается.

найти операцию: Найдите значение, указанное ByteBuf. Самый простой способ — использовать метод indexOf(). Более сложный метод можно получить, используя ByteBufProcessor в качестве параметра, напримерint index = buffer.forEachByte(ByteProcessor.FIND_CR);.

Производный буфер: Производный буфер обеспечивает представление доступа к ByteBuf. Представления обеспечивают только операцию доступа и не выполняют никаких операций копирования. Если вы измените определенное содержимое этого нового экземпляра ByteBuf, соответствующий исходный экземпляр также будет изменен.Если вам нужно скопировать настоящую копию существующего буфера, используйте методы copy() или copy(int, int).

операции чтения/записи: есть две категории чтения и записи:

Операции get() и set() — начинаются с заданного индекса и сохраняют индекс неизменным.
Операции чтения () и записи () — начинаются с заданного индекса и получают доступ к индексу на основе количества байтов, к которым был получен доступ.

Интерфейс ByteBufHolder

ByteBufHolder — это расширенная функция netty, обеспечивающая поддержку объединения буферов. Вы можете реализовать интерфейс ByteBufHolder через подклассы и добавлять нужные поля данных в соответствии с вашими потребностями. Может использоваться для полей расширения настраиваемого типа буфера.

Распределение ByetBuf

Распределить по требованию

Netty реализует пул (ByteBuf) через интерфейс ByteBufAllocator. Ссылку на ByteBufAllocator можно получить через Channel или ChannelHandler ChannelHandlerContext.

Netty предоставляет две реализации ByteBufAllocator: PooledByteBufAllocator (объединение) и UnpolledByteBufAllocator (не объединение). Netty по умолчанию использует PooledByteBufAllocator, но его также можно изменить с помощью ChannelConfig.

Необъединенный буфер

В случае отсутствия ссылки на ByteBufAllocator мы можем использовать класс инструментов Unpooled, предоставленный netty, для создания не объединенного ByteBufAllocator.

Класс ByteBufUtil

Класс ByteBufUtil предоставляет статические вспомогательные методы для управления ByteBuf.hexdump()Метод больше, чем содержимое ByteBuf в шестнадцатеричном представлении.equals(ByteBuf, ByteBuf)Метод используется для определения того, равны ли два экземпляра ByteBuf.

подсчет ссылок

netty представляет технологию подсчета ссылок для ButeBuf и ButeBufHolder, реализуя интерфейс ReferenceCounted. Когда счетчик равен 0, система освобождает буфер, что снижает накладные расходы на выделение памяти.