1

我是新来的风暴我在我的 spout 中使用 rabbitmq 从某个队列接收元组，并且有一个客户端正在运行另一台将元组插入该队列的机器我运行了一个简单的 rabbitmq 示例程序，它工作正常但是当我使用它时在风暴喷口内，它被阻挡在

connection = factory.newConnection();

即使我的rabbitmq服务器也在运行并且在我运行示例代码时在同一台机器上运行成功。打印语句 打印到语句

System.out.println(" setting host to 192.168.8.218..... ");

下面是我完整的 spout 类。

package storm.starter.spout;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import java.util.Map;
import java.util.Random;
import java.net.*;
import java.io.*;
import java.lang.Exception;
import java.io.IOException;

public class RabbitmqSpout extends BaseRichSpout {
    SpoutOutputCollector _collector;
    public final static String QUEUE_NAME = "record";
    ConnectionFactory factory;
    Connection connection;
    Channel channel;
    QueueingConsumer consumer;

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) 
    {
        _collector = collector;
        System.out.println(" [*] Intilization of spout..... ");

        try
        {
            factory = new ConnectionFactory();
            System.out.println(" creating connection factory..... ");
            factory.setHost("192.168.8.96");
            System.out.println(" setting host to 192.168.8.218..... ");
            connection = factory.newConnection();
            System.out.println(" creating new connection..... ");
            channel = connection.createChannel();
            System.out.println(" creating new channel..... ");
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println(" Declaring queue..... ");
            System.out.println(" [*] Waiting for messages. ");
        }
        catch(Exception exception)
        {
            System.out.println("Exception occurred. "+exception.getMessage());
        }

    }

    @Override
    public void nextTuple() 
    {
        System.out.println("In wait of tuples.... ");
        try
        {
            consumer = new QueueingConsumer(channel);
            System.out.println(" trying to consume..... ");
            channel.basicConsume(QUEUE_NAME, true, consumer);

            while (true) 
            {
                System.out.println(" trying to deliver..... ");
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" getting string..... ");
                System.out.println(" [x] Received '" + message + "'");
                System.out.print("emitting Rabbitmq Queue tuple");
                _collector.emit(new Values(message));
                System.out.print("emitted Rabbitmq Queue tuple");
            }   
        }

        catch(IOException io)
        {
            System.out.println("Exception occurred. ");
        }
        catch(Exception exception)
        {
            System.out.println("Exception occurred. ");
        }


    }        

    @Override
    public void ack(Object id) {
    }

    @Override
    public void fail(Object id) 
    {

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) 
    {
        declarer.declare(new Fields("record"));
    }

}

4

1 回答 1

0 
connection = factory.newConnection()

在RabbitMQConsumer()RabbitMQConsumer 类的方法中完成，因为数据由 RabbitMQ 消费者从队列中取出，然后将数据提供给您的 RabbitMQ spout。请参考：[ https://github.com/ppat/storm-rabbitmq/tree/master/src/main/java/io/latent/storm/rabbitmq][1]

于 2014-07-20T06:59:32.400 回答