使用RabbitMQ通过从队列中消费消息获取持久套接字连接

问题描述

我们想与客户端应用程序建立持久的套接字连接，我们要从rabbitmq发送数据。套接字连接设置keepAlive为真，但它仍然在每次交付时重置连接。 RabbitMQ 队列中的所有消息将在单套接字连接中发送而不关闭它。

ConnectionFactory factory = null;
        Connection connection = null;
        Channel channel = null;
        DefaultConsumer consumer = null;
        
        try {
            factory = new ConnectionFactory();
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setHost(hostname);
            connection = factory.newConnection();
            channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME,true,false,null);
            consumer = new DefaultConsumer(channel) {
            @Overrride
            public void handleDelivery(String consumerTag,Envelop envelop,AMQP.BasiscProperties,bytes[] body) throws IOException {
                String message = new String(body,"UTF-8");
                message = message.replaceAll("\\","");
                logger.debug("CSV Recieved -"+message);
                
                mainSocket = new Socket(HOST,IP);
                mainSocket.setKeepAlive(true);
                String out = Send(message,soc);
                logger.debug("Response after packet sent: "+ out);
            }
        };
        channel.basicConsume(QUEUE_NAME,consumer);
    }catch(Exception e) {
        logger.error("Exception: ",e);
    }

}

public String send(String message,Socket soc) {
    String retStr =null;
    
    OutputStream os = null;
    try  {
        os = soc.getoutputStream();
        messageBytes = message.getBytes();
        logger.info("Sending request to Socket Server");
        for(int i=0;i<message.length();i++) {
            os.write(messageBytes[i]);
        }
        retStr = "SUCCESS";
    }catch(ConnectException e) {
        logger.error("Exception: ",e);
        try {
            Thread.sleep(10000);
        }catch(InterruptedException e){
          logger.error("Exception: ",e);
        }
    }catch(Exception e) {
        logger.error("Exception: ",e);
    } finally {
        try {
            os.close();
        }catch(IOException e) {
            logger.error("Finally Exception: ",e);
        }
    }
    return retStr;
}

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）