Java 类com.rabbitmq.client.ConsumerCancelledException 实例源码

/**
 * @param args
 * @throws TimeoutException
 * @throws IOException
 * @throws InterruptedException
 * @throws ConsumerCancelledException
 * @throws ShutdownSignalException
 */
public static void main(String[] args) throws IOException, TimeoutException, ShutdownSignalException,
        ConsumerCancelledException, InterruptedException {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost(host);
    factory.setUsername(username);
    factory.setPassword(password);
    factory.setPort(port);
    factory.setVirtualHost(virtualHost);
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();
    channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
    channel.basicQos(1);
    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

    System.out.println("RPCServer Awating RPC request");
    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        BasicProperties props = delivery.getProperties();
        BasicProperties replyProps = new AMQP.BasicProperties.Builder().correlationId(props.getCorrelationId())
                .build();

        String message = new String(delivery.getBody(), "UTF-8");
        int n = Integer.parseInt(message);

        System.out.println("RPCServer fib(" + message + ")");
        String response = "" + fib(n);
        channel.basicPublish("", props.getReplyTo(), replyProps, response.getBytes());
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
}

/***
 * getParam().getWaitTime() 指定消息池为空时的堵塞超时
 * 
 */
@Override
public String readOneMessage()
{
  try
  {
    channel.basicConsume(getParam().getQueue(), false, consumer);
    QueueingConsumer.Delivery delivery = consumer.nextDelivery(getParam().getWaitTime());
    if (delivery != null)
    {
      deliveryTag = delivery != null ? delivery.getEnvelope().getDeliveryTag() : deliveryTag;
      String msg = getMessageContent(delivery);
      return msg;
    }
    else
      return null;
  }
  catch (IOException | ShutdownSignalException | ConsumerCancelledException | InterruptedException e)
  {
    throw new MqReceiveException(e);
  }
}

/**
 * Retrieves the benchmarks registered at the HOBBIT PlatformController
 *
 * @return A list of benchmarks
 * @throws IOException
 * @throws InterruptedException
 * @throws ConsumerCancelledException
 * @throws ShutdownSignalException
 *             If something goes wrong with the request
 */
public List<BenchmarkBean> requestBenchmarks()
        throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
    LOGGER.info("Sending request...");
    byte[] data = client.request(new byte[] { FrontEndApiCommands.LIST_AVAILABLE_BENCHMARKS });
    if (data == null) {
        throw new IOException("Didn't got a response.");
    }

    LOGGER.info("Parsing response...");
    // parse the response
    String jsonString = RabbitMQUtils.readString(data);
    Collection<BenchmarkMetaData> benchmarks = gson.fromJson(jsonString,
            new TypeToken<Collection<BenchmarkMetaData>>() {
            }.getType());

    LOGGER.info("Preparing response for GUI...");
    // Create output
    List<BenchmarkBean> benchmarkBeans = new ArrayList<BenchmarkBean>();

    for (BenchmarkMetaData benchmark : benchmarks) {
        benchmarkBeans.add(
                new BenchmarkBean(benchmark.benchmarkUri, benchmark.benchmarkName, benchmark.benchmarkDescription));
    }

    LOGGER.debug(Arrays.toString(benchmarkBeans.toArray()));
    LOGGER.info("Sending response to GUI...");

    return benchmarkBeans;
}

public void begin() throws IOException, TimeoutException, ShutdownSignalException, 
                    ConsumerCancelledException, InterruptedException, SpiderSettingFileException{
    readSetting();
    logger.info("worker [" + this.settingObject.getWorkerid() + "] start...");
    for (Entry<String, MQRecver> recv : this.recvfromMap.entrySet()) {
        new Thread(new RecvThread(this, recv.getKey(), recv.getValue(), this.sendtoMap)).start();
    }
}

public RabbitPubSubMsg consume() throws ShutdownSignalException, ConsumerCancelledException, InterruptedException {     
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        String message = new String(delivery.getBody());
        String routingKey = delivery.getEnvelope().getRoutingKey();
        System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");
        RabbitPubSubMsg ret = new RabbitPubSubMsg(message, routingKey);
        return ret;
}

private static void rabbitMQEventListener(String QUEUE_NAME, String eventIdentifier) {
  try {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    com.rabbitmq.client.Connection rabbitmqConnection;
    rabbitmqConnection = factory.newConnection();
    Channel channel = rabbitmqConnection.createChannel();
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    System.out.println(" [*] "+QUEUE_NAME+" server is waiting for messages. To exit press CTRL+C");

    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicConsume(QUEUE_NAME, true, consumer);
    while (true) {
      QueueingConsumer.Delivery delivery = null;
      try {
        delivery = consumer.nextDelivery();
      } catch (ShutdownSignalException | ConsumerCancelledException
          | InterruptedException e) {
        e.printStackTrace();
      }
      String message = new String(delivery.getBody());
      System.out.println(" [...x...] "+QUEUE_NAME+" server received '" + message + "'");

      server.getBroadcastOperations().sendEvent(eventIdentifier, message);

    }
  } catch (IOException e1) {
    e1.printStackTrace();
  }
}

@Override
public void run() {
  try {
    ObjectMapper mapper = new ObjectMapper();
    Entry<Channel, QueueingConsumer> rabbitmq = RabbitMQ.createChannelConsumer(TASK_QUEUE_NAME, ident);
    Channel channel = rabbitmq.getKey();
    QueueingConsumer consumer = rabbitmq.getValue();

    System.out.println(this.ident + " Ready for RPC requests on " + TASK_QUEUE_NAME);

    while (true) {
      // Get a message (task) from an endpoint
      QueueingConsumer.Delivery delivery;
      delivery = consumer.nextDelivery();
      AMQP.BasicProperties props = delivery.getProperties();
      AMQP.BasicProperties.Builder replyProps = new AMQP.BasicProperties.Builder().correlationId(props.getCorrelationId());
      byte[] response;

      String endpoint = new String(delivery.getBody());

      RateLimitStatusImpl rl = new RateLimitStatusImpl();

      for (TwitterWorker w : BOTS.get(endpoint.toString())) {
        rl = rl.mergeWith(w.cachedRateLimit);
      }

      response = mapper.writeValueAsBytes(rl);
      channel.basicPublish("", props.getReplyTo(), replyProps.build(), response);
      channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

      System.out.println(this.ident + " " + endpoint + " " + rl.getRemaining() + " of " + rl.getLimit() + " Reset in " + rl.getSecondsUntilReset());
    }

  } catch (IOException | ShutdownSignalException | ConsumerCancelledException | InterruptedException e) {
    e.printStackTrace();
  }
}

public static final void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException
{
    ConnectionFactory factory = new ConnectionFactory();
    factory.setVirtualHost("/");
    factory.setHost("localhost");
    factory.setUsername("guest");
    factory.setPassword("guest");
    factory.setPort(5672);
    ExecutorService executor = Executors.newFixedThreadPool(POOL_SIZE);
    AMQPRPCServer server = new AMQPRPCServer();
    for (int i = 0; i < POOL_SIZE; i++) {
      Runnable worker = server.new AMQPListner(server, factory, i);
      executor.execute(worker);
    }
}

/**
 * Retrieves the benchmark details from the HOBBIT PlatformControler
 *
 * @param benchmarkUri
 *            the URI of the benchmark for which the details should be retrieved
 * @param user
 *            information about the requesting user which will be used to filter
 *            the systems that can be used with the requested benchmark.
 * @return
 * @throws GUIBackendException
 * @throws IOException
 * @throws InterruptedException
 * @throws ConsumerCancelledException
 * @throws ShutdownSignalException
 */
public BenchmarkBean requestBenchmarkDetails(String benchmarkUri, UserInfoBean user) throws GUIBackendException,
        IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
    LOGGER.info("Sending request...");
    // Map<String, String> env = System.getenv();
    if (benchmarkUri == null) {
        String msg = "Benchmark URI is null. Aborting.";
        LOGGER.error(msg);
        throw new GUIBackendException(msg);
    }
    LOGGER.info("Sending request...");

    byte[] data = null;
    if (user != null) {
        data = client.request(RabbitMQUtils.writeByteArrays(
                new byte[] { FrontEndApiCommands.GET_BENCHMARK_DETAILS }, new byte[][] {
                        RabbitMQUtils.writeString(benchmarkUri), RabbitMQUtils.writeString(user.getEmail()) },
                null));
    } else {
        data = client
                .request(RabbitMQUtils.writeByteArrays(new byte[] { FrontEndApiCommands.GET_BENCHMARK_DETAILS },
                        new byte[][] { RabbitMQUtils.writeString(benchmarkUri) }, null));
    }
    if (data == null) {
        throw new IOException("Didn't got a response.");
    }

    Model benchmarkModel = null;
    Collection<SystemMetaData> systems = null;
    try {
        LOGGER.info("Parsing response...");
        // parse the response
        ByteBuffer buffer = ByteBuffer.wrap(data);
        benchmarkModel = RabbitMQUtils.readModel(buffer);
        String jsonString = RabbitMQUtils.readString(buffer);
        systems = gson.fromJson(jsonString, new TypeToken<Collection<SystemMetaData>>() {
        }.getType());
    } catch (Exception e) {
        throw new IOException("Error while parsing benchmark model.", e);
    }

    BenchmarkBean benchmarkDetails = RdfModelHelper.createBenchmarkBean(benchmarkModel);
    if (benchmarkDetails == null) {
        throw new IOException("Error while parsing benchmark model.");
    }

    // Parse Benchmark System Details
    LOGGER.info("Adding systems for GUI...");
    benchmarkDetails.setSystems(new ArrayList<>());
    if (systems != null) {
        for (SystemMetaData system : systems) {
            benchmarkDetails.getSystems()
                    .add(new SystemBean(system.systemUri, system.systemName, system.systemDescription));
        }
    }

    LOGGER.info("Sending response to GUI...");
    return benchmarkDetails;
}

public MQItem recv() throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    MQItem item = gson.fromJson(new String(delivery.getBody()), MQItem.class);
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    return item;
}

public String consume() throws ShutdownSignalException, ConsumerCancelledException, InterruptedException {
    String ret = null;
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    ret = new String(delivery.getBody());
    return ret;
}

public byte[] receive() throws ShutdownSignalException, ConsumerCancelledException, InterruptedException {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    return delivery.getBody();
}

public byte[] receive() throws ShutdownSignalException, ConsumerCancelledException, InterruptedException {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    return delivery.getBody();
}

public Delivery nextDelivery() throws InterruptedException, ShutdownSignalException, ConsumerCancelledException
{
    return queue.take();
}

public Delivery nextDelivery(long timeout) throws InterruptedException, ShutdownSignalException, ConsumerCancelledException
{
    return queue.poll(timeout, TimeUnit.MILLISECONDS);
}

@Override
public void handleCancel(String consumerTag) throws IOException {
    cancelled = new ConsumerCancelledException();
}

public static boolean shouldReconnect(Throwable e) {
    if(e == null)
        return false;

    if(e instanceof ConnectException) {
        //rabbit server is not up
        return true;
    }

    if(e instanceof ShutdownSignalException) {
        Object reason = ((ShutdownSignalException) e).getReason();
        if(reason instanceof AMQCommand) {
            com.rabbitmq.client.Method method = ((AMQCommand) reason).getMethod();
            if(method instanceof Channel.Close) {
                if(((Channel.Close) method).getReplyCode() == AMQP.PRECONDITION_FAILED) {
                    //this is the case when we declare a queue with different params than the one that exists
                    //TODO: have a configuration option to ignore the erroring method call and continue on
                    //      if the queue exists... who cares?
                    return false;
                }
            }
        }

        /*
         * If we didn't initiate the shutdown, or we encounter a connection error... we should reconnect
         * isHardError returns true if it was a connection error... false if a channel error
         */
        return !((ShutdownSignalException) e).isInitiatedByApplication();
            /*((ShutdownSignalException)e).isHardError()*/
    }

    if(e instanceof ConsumerCancelledException) {
        //if the server cancelled our consumer, we should reconnect
        return true;
    }

    if(e instanceof IOException) {
        return shouldReconnect(e.getCause());           
    }

    return false;
}

/**
 * Main application-side API: wait for the next message delivery and return it.
 * @return the next message
 * @throws InterruptedException if an interrupt is received while waiting
 * @throws ShutdownSignalException if the connection is shut down while waiting
 * @throws ConsumerCancelledException if this consumer is cancelled while waiting
 */
public HaDelivery nextDelivery()
    throws InterruptedException, ShutdownSignalException, ConsumerCancelledException
{
    return handle(queue.take());
}

/**
 * Main application-side API: wait for the next message delivery and return it.
 * @param timeout timeout in millisecond
 * @return the next message or null if timed out
 * @throws InterruptedException if an interrupt is received while waiting
 * @throws ShutdownSignalException if the connection is shut down while waiting
 * @throws ConsumerCancelledException if this consumer is cancelled while waiting
 */
public HaDelivery nextDelivery(long timeout)
    throws InterruptedException, ShutdownSignalException, ConsumerCancelledException
{
    return handle(queue.poll(timeout, TimeUnit.MILLISECONDS));
}