Shulou(Shulou.com)06/02 Report--

Handler on netty's pipeline processing chain: requires IdleStateHandler heartbeat to detect whether channel is OK, as well as UserAuthHandler to handle login authentication and message processing MessageHandler

Protected void initChannel (SocketChannel ch) throws Exception {ch.pipeline () .addLast (defLoopGroup, / / codec new HttpServerCodec (), / / convert multiple messages into a single message object new HttpObjectAggregator (65536), / / support asynchronous transmission of large bitstreams Generally used to send file stream new ChunkedWriteHandler (), / / check whether the link is read idle, cooperate with heartbeat handler to check whether channel is normal new IdleStateHandler (60,0,0), / / handle handshake and authentication new UserAuthHandler (), / / process message sending new MessageHandler () }

For all connected channel, we need to save them, and we need to rely on these channel for future mass messages.

Public static void addChannel (Channel channel) {String remoteAddr = NettyUtil.parseChannelRemoteAddr (channel); System.out.println ("addChannel:" + remoteAddr); if (! channel.isActive ()) {logger.error ("channel is not active, address: {}", remoteAddr);} UserInfo userInfo = new UserInfo (); userInfo.setAddr (remoteAddr); userInfo.setChannel (channel) UserInfo.setTime (System.currentTimeMillis ()); userInfos.put (channel, userInfo);}

After logging in, channel becomes a valid channel, and invalid channel will be discarded after that

Public static boolean saveUser (Channel channel, String nick, String password) {UserInfo userInfo = userInfos.get (channel); if (userInfo = = null) {return false;} if (! channel.isActive ()) {logger.error ("channel is not active, address: {}, nick: {}", userInfo.getAddr (), nick); return false } if (nick = = null | | password = = null) {return false;} LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper (); lambdaQueryWrapper.eq (Account::getUsername, nick) .eq (Account::getPassword, password); Account account = accountMapperStatic.selectOne (lambdaQueryWrapper); if (account = = null) {return false } / / add an authenticated user userCount.incrementAndGet (); userInfo.setNick (nick); userInfo.setAuth (true); userInfo.setId (account.getId ()); userInfo.setUsername (account.getUsername ()); userInfo.setGroupNumber (account.getGroupNumber ()); userInfo.setTime (System.currentTimeMillis ()) / / register the channel offlineInfoTransmitStatic.registerPull (channel) of the user's push message; return true;}

When channel is closed, messages are no longer received. UnregisterPull is the consumer of logged-out information, and the client no longer receives chat messages. In addition, there is a write lock operation from below to avoid suddenly shutting down channel while channel is still sending messages, which will lead to an error.

Public static void removeChannel (Channel channel) {try {logger.warn ("channel will be remove, address is: {}", NettyUtil.parseChannelRemoteAddr (channel)); / / add a read-write lock to ensure that the channel is removed to avoid other threads operating on it when channel is closed, resulting in an error rwLock.writeLock (). Lock (); channel.close () UserInfo userInfo = userInfos.get (channel); if (userInfo! = null) {if (userInfo.isAuth ()) {offlineInfoTransmitStatic.unregisterPull (channel); / / minus an authenticated user userCount.decrementAndGet ();} userInfos.remove (channel) }} finally {rwLock.writeLock () .unlock ();}}

In order to seamlessly switch the four states of using rabbitmq, rocketmq, activemq and not using middleware to store and forward chat messages, the following four interfaces are defined. The order is to send single chat message, group chat message, the client starts to receive the message, and the client goes offline and does not receive the message.

Public interface OfflineInfoTransmit {void pushP2P (Integer userId, String message); void pushGroup (String groupNumber, String message); void registerPull (Channel channel); void unregisterPull (Channel channel);}

Among them, how to use one of rabbitmq, rocketmq, activemq middleware to store and forward chat messages, its processing flow is as follows:

The model of a single chat refers to the model of the thread pool, and if the user is online, it is sent directly to the user through channel. If the user is offline, it will be sent to the middleware storage, and the next time the user goes online, the message will be pulled directly from the middleware. Compared with all messages sent through middleware, the advantage of this is that the performance of group chat is improved, while group chat is completely forwarded through middleware, message sending middleware, and the client receives messages from middleware. If you still operate like a single chat, online users send directly through channel, which is too tedious to determine which users in this group are online. If the user is online, register the consumer and receive the message from the middleware. Otherwise, the consumer is disconnected and the message is kept in the middleware so that the client can pull it the next time it comes online. In this way, the reception of offline messages is realized. No matter which kind of middleware is used or whether or not it uses middleware, its processing flow follows the above three requirements, so it can seamlessly switch between the above four methods to store and forward messages. Which method is needed to open the corresponding comments.

Code address:

Https://github.com/shuangyueliao/netty-chat

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.