在 HornetQ 中已与 NettyAcceptor 一起使用的地址

Question

我正在开发一个将 HotnetQ 嵌入到 pubsub 服务的 Spring 应用程序服务器。我有下一节课，下面将详细介绍给管理 HornetQ 服务器。

但是，当我部署我的应用程序时，我在 tomcat 控制台中收到下一个错误。

我一直在更改 Netty 接受器端口号，但问题仍然存在，如果我使用其他接受器而不是 Netty，它工作正常，但我对 Netty 接受器类型感兴趣。

Setting ServerMQ configuration...
Initializing ServerMQ...
mar 12, 2014 1:42:44 PM org.hornetq.core.server.impl.HornetQServerImpl start
INFO: HQ221000: live server is starting with configuration HornetQ Configuration (clustered=false,backup=false,sharedStore=true,journalDirectory=data/journal,bindingsDirectory=data/bindings,largeMessagesDirectory=data/largemessages,pagingDirectory=data/paging)
mar 12, 2014 1:42:44 PM org.hornetq.core.server.impl.HornetQServerImpl initialisePart1
WARN: HQ222007: Security risk! HornetQ is running with the default cluster admin user and default password. Please see the HornetQ user guide, cluster chapter, for instructions on how to change this.
mar 12, 2014 1:42:44 PM org.hornetq.core.remoting.server.impl.RemotingServiceImpl <init>
INFO: HQ221043: Adding protocol support CORE
ServerMQ initialized!
mar 12, 2014 1:42:44 PM org.hornetq.core.server.impl.HornetQServerImpl$SharedNothingLiveActivation run
ERROR: HQ224000: Failure in initialisation
java.net.BindException: Address already in use: bind
    at sun.nio.ch.Net.bind0(Native Method)
    at sun.nio.ch.Net.bind(Net.java:444)
    at sun.nio.ch.Net.bind(Net.java:436)
    at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:214)
    at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:74)
    at io.netty.channel.socket.nio.NioServerSocketChannel.doBind(NioServerSocketChannel.java:102)
    at io.netty.channel.AbstractChannel$AbstractUnsafe.bind(AbstractChannel.java:479)
    at io.netty.channel.DefaultChannelPipeline$HeadHandler.bind(DefaultChannelPipeline.java:1000)
    at io.netty.channel.DefaultChannelHandlerContext.invokeBind(DefaultChannelHandlerContext.java:457)
    at io.netty.channel.DefaultChannelHandlerContext.bind(DefaultChannelHandlerContext.java:442)
    at io.netty.channel.DefaultChannelPipeline.bind(DefaultChannelPipeline.java:842)
    at io.netty.channel.AbstractChannel.bind(AbstractChannel.java:194)
    at io.netty.bootstrap.AbstractBootstrap$2.run(AbstractBootstrap.java:331)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:354)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:353)
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:101)
    at java.lang.Thread.run(Thread.java:744)

mar 12, 2014 1:42:44 PM org.hornetq.core.server.impl.HornetQServerImpl start
INFO: HQ221001: HornetQ Server version 2.4.1.Final (Fast Hornet, 124) [a681c9b2-a83a-11e3-b8ef-531df38c9cfa]

这是我的类实现。

@Component(value = "serverMQ")
public class ServerMQ extends EmbeddedHornetQ {

    // Server config.
    private Configuration config;

    private ClientSessionFactory clientSessionFactory;

    public ServerMQ() {

        System.out.println("Setting ServerMQ configuration...");

        // Instantiate server config.
        this.config = new ConfigurationImpl();

        // Server has persistence for messages.
        this.config.setPersistenceEnabled(false);

        // Server has user security authentication.
        this.config.setSecurityEnabled(false);

        Map<String, Object> nettyAcceptorParams = new HashMap<String, Object>();
        nettyAcceptorParams.put(TransportConstants.HOST_PROP_NAME, "localhost");
        nettyAcceptorParams.put(TransportConstants.PORT_PROP_NAME, 5555);

        this.config.getAcceptorConfigurations().clear();

        this.config.getAcceptorConfigurations().add(new TransportConfiguration(NettyAcceptorFactory.class.getName(), nettyAcceptorParams));
    }

    @PostConstruct
    public void init() {

        try {

            System.out.println("Initializing ServerMQ...");

            this.setConfiguration(this.config);
            this.start();

            System.out.println("ServerMQ initialized!");
        } catch (Exception ex) {

            System.err.println("ServerMQ initializing error:\n" + ex.getMessage());
        }
    }

}

在 @Clebert 建议之后，我更改了我的 ServerMQ 类以尝试新的方式来实现 pubsub 服务，如下所示...同样的问题... p>

@Component(value = "serverMQ")
public class ServerMQ {

    // Server config.
    private Configuration config;

    // HornetQ Server
    private EmbeddedHornetQ mQServer;

    private ClientSessionFactory clientSessionFactory;
public ServerMQ() {

    System.out.println("Setting ServerMQ configuration...");

    // Instantiate server config.
    this.config = new ConfigurationImpl();

    // Server has persistence for messages.
    this.config.setPersistenceEnabled(false);

    // Server has user security authentication.
    this.config.setSecurityEnabled(false);

    Map<String, Object> nettyAcceptorAttrs = new HashMap<String, Object>();
    nettyAcceptorAttrs.put(TransportConstants.HOST_PROP_NAME, "localhost");
    nettyAcceptorAttrs.put(TransportConstants.PORT_PROP_NAME, 5555);

    /*
     HashSet<TransportConfiguration> transports = new HashSet<TransportConfiguration>();
     transports.add(new TransportConfiguration(NettyAcceptorFactory.class.getName(), nettyAcceptorParams));
     transports.add(new TransportConfiguration(InVMAcceptorFactory.class.getName()));
     */

    this.config.getAcceptorConfigurations().clear();

    this.config.getAcceptorConfigurations().add(new TransportConfiguration(NettyAcceptorFactory.class.getName(), nettyAcceptorAttrs));

    this.init2();
}

public void init2() {

    try {

        System.out.println("Initializing ServerMQ...");

        this.mQServer = new EmbeddedHornetQ();

        this.mQServer.setConfiguration(this.config);
        this.mQServer.start();
        /*
         this.clientSessionFactory = HornetQClient.createServerLocatorWithoutHA(
         new TransportConfiguration(
         InVMConnectorFactory.class.getName())).createSessionFactory();
         */
        System.out.println("ServerMQ initialized!");
    } catch (Exception ex) {

        System.err.println("ServerMQ initializing error:\n" + ex.getMessage());
    }
}

}

Answer 1

您正在启动两台服务器！

您的课程正在扩展 EmbeddedHornetQ，它还启动了一个 hornetQServer..

在您的子类上，您将启动一个新服务器。

不同之处在于，超类上的 EmbeddedServer 使用文件配置解析器和 xml，而您的版本则从头开始。

修复你的代码，只启动一个服务器！

在您的第二次迭代之后，我将您的代码作为主类运行，它在这里工作。

package org.hornetq.tests.integration;

import java.util.HashMap;
import java.util.Map;

import org.hornetq.api.core.TransportConfiguration;
import org.hornetq.api.core.client.ClientSessionFactory;
import org.hornetq.core.config.Configuration;
import org.hornetq.core.config.impl.ConfigurationImpl;
import org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory;
import org.hornetq.core.remoting.impl.netty.TransportConstants;
import org.hornetq.core.server.embedded.EmbeddedHornetQ;

public class ServerMQ
{

   // Server config.
   private Configuration config;

   // HornetQ Server
   private EmbeddedHornetQ mQServer;

   private ClientSessionFactory clientSessionFactory;

   public ServerMQ()
   {

      System.out.println("Setting ServerMQ configuration...");

      // Instantiate server config.
      this.config = new ConfigurationImpl();

      // Server has persistence for messages.
      this.config.setPersistenceEnabled(false);

      // Server has user security authentication.
      this.config.setSecurityEnabled(false);

      Map<String, Object> nettyAcceptorAttrs = new HashMap<String, Object>();
      nettyAcceptorAttrs.put(TransportConstants.HOST_PROP_NAME, "localhost");
      nettyAcceptorAttrs.put(TransportConstants.PORT_PROP_NAME, 5555);

    /*
     HashSet<TransportConfiguration> transports = new HashSet<TransportConfiguration>();
     transports.add(new TransportConfiguration(NettyAcceptorFactory.class.getName(), nettyAcceptorParams));
     transports.add(new TransportConfiguration(InVMAcceptorFactory.class.getName()));
     */

      this.config.getAcceptorConfigurations().clear();

      this.config.getAcceptorConfigurations().add(new TransportConfiguration(NettyAcceptorFactory.class.getName(), nettyAcceptorAttrs));

      this.init2();
   }

   public void init2()
   {

      try
      {

         System.out.println("Initializing ServerMQ...");

         this.mQServer = new EmbeddedHornetQ();

         this.mQServer.setConfiguration(this.config);
         this.mQServer.start();
        /*
         this.clientSessionFactory = HornetQClient.createServerLocatorWithoutHA(
         new TransportConfiguration(
         InVMConnectorFactory.class.getName())).createSessionFactory();
         */
         System.out.println("ServerMQ initialized!");
      }
      catch (Exception ex)
      {

         System.err.println("ServerMQ initializing error:\n" + ex.getMessage());
      }
   }

主类

   public static void main(String arg[])
   {
      ServerMQ mq = new ServerMQ();

      try
      {
         Thread.sleep(10000);
      }
      catch (Exception e)
      {
         e.printStackTrace();
      }
   }

}

您现在有 spring 为您的 init 调用...您正在启动服务器两次..或者您的 xml 上有两个配置正在从 Embedded 执行。

或者您的系统中某处可能有一个死进程。尝试重新启动您的服务器.. 或检查是否有任何其他使用您配置的端口。