spring-boot 整合 netty与rabbitMQ 处理不能消费的问题

1、 在集成 netty 与rabbitMQ 时，发现 不收接收 rabbitMQ消息。

具体不能监听消息通道的代码如下  我是监听ApplicationListener的时候 启动的netty服务。这样不能加入mq消息通道，思考这里是不是MQbean没有注入呢。所以我就修改了下注入方式。

package com.wawi.imsocket;
 
 
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.wawi.imsocket.socket.SocketPipelineAdd;
import com.wawi.imsocket.utls.NetWorkUtils;
import com.wawi.imsocket.utls.SpringUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;
 
@Slf4j
@Component
public class NettyServerBootstrap implements ApplicationListener<ContextRefreshedEvent> {
 
    @Value("${nettyWebsocket.port:10025}")
    private int socketServerPort;
    @Value("${nettyWebsocket.name:wc-service}")
    private String name;
    @Value("${nettyWebsocket.nacosAddress:127.0.0.1}")
    private String nacosAddress;
    @Value("${nettyWebsocket.nacosPort:8848}")
    private Integer nacosPort;
    /**
     * 启动对应服务器
     */
    public void startup() {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.option(ChannelOption.SO_BACKLOG, 4096);
            b.group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            SpringUtil.getBean(SocketPipelineAdd.class).handAdd(ch.pipeline());
                        }
                    });
            ChannelFuture future = b.bind("0.0.0.0", socketServerPort).sync();
            log.info("================启动成功，端口号：{}========", socketServerPort);
            init();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            log.info("server exit...");
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
 
    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        SpringUtil.getBean(NettyServerBootstrap.class).startup();
    }
    public void init() {
        try {
            //获取nacos服务
            NamingService namingService = NamingFactory.createNamingService(nacosAddress+":" +nacosPort);
            //将服务注册到注册中心\
//            System.out.println(NetWorkUtils.getHost());
            namingService.registerInstance(name, NetWorkUtils.getHost(), Integer.valueOf(socketServerPort));
        } catch (Exception e) {
            e.printStackTrace();
            log.error("注册nacos失败", e);
        }
    }
}

修改如下 取消 启动 监听 直接在启动main 之后直接启动 

package com.wawi.imsocket;
 
 
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.wawi.imsocket.socket.SocketPipelineAdd;
import com.wawi.imsocket.utls.NetWorkUtils;
import com.wawi.imsocket.utls.SpringUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;
 
@Slf4j
@Component
public class NettyServerBootstrap  {
 
    @Value("${nettyWebsocket.port:10025}")
    private int socketServerPort;
    @Value("${nettyWebsocket.name:wc-service}")
    private String name;
    @Value("${nettyWebsocket.nacosAddress:127.0.0.1}")
    private String nacosAddress;
    @Value("${nettyWebsocket.nacosPort:8848}")
    private Integer nacosPort;
    /**
     * 启动对应服务器
     */
    public void startup() {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.option(ChannelOption.SO_BACKLOG, 4096);
            b.group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            SpringUtil.getBean(SocketPipelineAdd.class).handAdd(ch.pipeline());
                        }
                    });
            ChannelFuture future = b.bind("0.0.0.0", socketServerPort).sync();
            log.info("================启动成功，端口号：{}========", socketServerPort);
            init();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            log.info("server exit...");
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
    
    public void init() {
        try {
            //获取nacos服务
            NamingService namingService = NamingFactory.createNamingService(nacosAddress+":" +nacosPort);
            //将服务注册到注册中心\
//            System.out.println(NetWorkUtils.getHost());
            namingService.registerInstance(name, NetWorkUtils.getHost(), Integer.valueOf(socketServerPort));
        } catch (Exception e) {
            e.printStackTrace();
            log.error("注册nacos失败", e);
        }
    }
}

 这里是重点 在 run之后启动 netty服务，这样就能创建 mq queue了

@SpringBootApplication
@EnableDiscoveryClient
@EnableCircuitBreaker
@ComponentScan(basePackages= {"com.wawi","cn.wawi"})
@EnableCaching
public class ImSocketApplication {
   @Bean
   public SpringUtil getSpringUtil() {
      return new SpringUtil();
   }
   public static void main(String[] args) {
      SpringApplication.run(ImSocketApplication.class, args);
      SpringUtil.getBean(NettyServerBootstrap.class).startup();
   }
 
}