camel与activemq搭建netty4框架

矫情吗；* 2022-06-11 01:54 130阅读 0赞

摘要：  
笔者现在参与的项目是有设备或安全保往服务器发送数据，APP端/WEB端请求服务器的数据，笔者感觉单独的netty4服务器转发起来太麻烦，累，所以就尝试一下用camel和activemq搭建netty4服务，别忘了，你还要搭起activemq服务器，当然了，serviceMix自带了activemq，初学阶段，一起讨论学习.  
camel学习可以参考[http://camel.apache.org/netty4.html][http_camel.apache.org_netty4.html]，  
Activemq学习参考[http://activemq.apache.org/][http_activemq.apache.org]  
serviceMix学习参考[http://servicemix.apache.org/][http_servicemix.apache.org]  
项目环境：OSGI+serviceMix+maven  
有些朋项目的环境和我不一样的，自己就看着改吧，下面我们开始写  
**一.依赖的jar**  
这里就只把camel/activemq/netty依赖的jar贴出来了

<dependency>
            <groupId>javax.inject</groupId>
            <artifactId>javax.inject</artifactId>
            <version>1</version>
            <optional>true</optional>
           </dependency>
    
           <dependency>
                <groupId>org.apache.camel</groupId>
                <artifactId>camel-netty4</artifactId>
                <version>2.16.4</version>
                <scope>provided</scope>
            </dependency>
    
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>5.14.5</version>
            </dependency>

**二.项目结构**

![这里写图片描述][SouthEast]

**三.代码块**  
我们先看一下解码器的工厂

import io.netty.channel.ChannelHandler;
    import org.apache.camel.component.netty4.DefaultChannelHandlerFactory;
    
    import javax.inject.Named;
    
    @Named
    public class DecoderFactory extends DefaultChannelHandlerFactory { 
        @Override
        public ChannelHandler newChannelHandler() {
            return new Decoder();
        }
    }

解码器  
特别提醒handlerRemoved0()方法和channelRegistered()方法，可以对在线和离线的客户端做监听

import com.fasterxml.jackson.databind.ObjectMapper;
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ByteToMessageDecoder;
    
    import java.util.List;
    
    public class Decoder extends ByteToMessageDecoder { 
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) throws Exception {
            byte[] buff = new byte[buf.readableBytes()];
            buf.readBytes(buff);
            //转换为字符串
            String message=new String(buff,"utf-8");
            ObjectMapper objectMapper=new ObjectMapper();
    
            //转换为json格式的字符串 实体类亦可这样转
            String json = objectMapper.writeValueAsString(message);
            //发送消息到队列(topic)
            out.add(json);
    
            //响应客户端请求的
            ctx.writeAndFlush("hello client");
        }
    
        @Override
        protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception {
            super.handlerRemoved0(ctx);
            //客户端断开连接时，触发此方法
        }
    
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            super.channelRegistered(ctx);
            //客户端连接成功时，触发此方法
        }
    }

接下来是编码器工厂

import io.netty.channel.ChannelHandler;
    import org.apache.camel.component.netty4.DefaultChannelHandlerFactory;
    
    import javax.inject.Named;
    
    @Named
    public class EncoderFactory extends DefaultChannelHandlerFactory { 
    
        @Override
        public ChannelHandler newChannelHandler() {
            return new Encoder();
        }
    }

编码器  
这里要提醒的是，这个泛型是看解码器里返回数据的类型，我这里为了简单直接返回字符串，业务复杂一点的基本是返回一个对象到编码器里，编码器里又可以返回各种类型到客户端

public class Encoder extends MessageToByteEncoder<String> { 
    
        @Override
        protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
            //直接发送响应客户端的消息
            out.writeBytes(msg.getBytes());
        }
    }

好，接下来，我们就来看看配置文件了，因为我的是用了osgi技术的，配置文件放在resource下面，文件命名按我上面图上所示，文件就会自动读到，在这里我们netty4放出的端口是46680

<?xml version="1.0" encoding="UTF-8"?>
    <blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0" default-timeout="0">
    
        <camelContext xmlns="http://camel.apache.org/schema/blueprint" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://camel.apache.org/schema/blueprint http://camel.apache.org/schema/blueprint/camel-blueprint.xsd">
            <route>
                <from uri="netty4:tcp://0.0.0.0:46680?sync=true&decoders=#decoderFactory&encoders=#encoderFactory" />
                <to uri="activemq:topic:testNetty?username=admin&password=admin" />
            </route>
    
        </camelContext>
    
    </blueprint>

我们对上面的配置文件大概说一下  
其中decoderFactory和encoderFactory是上面@Named注解生成出来的，生成的效果如下图所示  
![这里写图片描述][SouthEast 1]

**四.测试**  
好，写完了，我们来做测试，在这里介绍两个测试工具MQTT.fx和PuTTY  
![这里写图片描述][SouthEast 2]

结果的如下图所示  
![这里写图片描述][SouthEast 3]  
![这里写图片描述][SouthEast 4]  
很明显，客户端发送的消息已经在testNetty队列里头了，相信大家已经明白了，只要哪个用户想看某一设备的数据，只要订阅服务器端的topic就OK了，这样转发数据就大大的减少了服务器的压力，也简单了好多！  
由于刚开始学着写博客，对这方面技术也没有深入了解，写的不好的请朋友们多多体谅

[http_camel.apache.org_netty4.html]: http://camel.apache.org/netty4.html
[http_activemq.apache.org]: http://activemq.apache.org/
[http_servicemix.apache.org]: http://servicemix.apache.org/
[SouthEast]: /images/20220611/ce7fe5dde33249688ca2592e4ac9b553.png
[SouthEast 1]: /images/20220611/c1bae5c70f32435c9178126563133f5f.png
[SouthEast 2]: /images/20220611/d5cf5f6ebacd4a7a85cc6854ce657344.png
[SouthEast 3]: /images/20220611/5a4a2f6ef9c5479bb50fb431ae8763ac.png
[SouthEast 4]: /images/20220611/e2896120966b4acb8bd8fb4a42577b00.png