Java 类com.rabbitmq.client.QueueingConsumer.Delivery 实例源码

public static void main(String[] args) throws Exception {

        ConnectionFactory factory = new ConnectionFactory();
//      setRemoteConnectionFactory(factory);
        setLocalConnectionFactory(factory);

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME, true, consumer);

        while (true) {
            Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" [x] Received '" + message + "'");
        }
    }

private void runListener(String queueName, final OutputStream os) throws Exception {
    if (queueName != null) {
        channel.queueDeclare(queueName, true, false, false, null);
    } else {
        queueName = channel.queueDeclare().getQueue();              
    }

    channel.queueBind(queueName, exchangeName, "");

    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicConsume(queueName, false, consumer);
    int c = 0;

    while (true) {
        Delivery d = consumer.nextDelivery(5000);
        c += 1;
        if (d != null) {
            os.write(d.getBody());
            channel.basicAck(d.getEnvelope().getDeliveryTag(), true);
        }
        if (d == null || c > 1000) {
            os.flush();
            c = 0;
        }
    }
}

public static void main(String args[]) throws Exception {
    if (args.length < 2) {
        System.err.println("Usage: " + RMQSpout.class.getCanonicalName() + " <rmquri> <exchange> [queuename] ");
        return;
    }

    RMQSpout rmq;
    if (args.length == 2) {
        rmq = new RMQSpout(args[0], args[1]);
    } else {
        rmq = new RMQSpout(args[0], args[3]);
    }
    rmq.connect();

    while (true) {
        Delivery d = rmq.consumer.nextDelivery();
        System.out.println(new String(d.getBody()));
        rmq.channel.basicAck(d.getEnvelope().getDeliveryTag(), false);
    }

}

@Override
public boolean hasNext() throws IOException, CollectionException {

    try {
        Delivery d = consumer.nextDelivery(timeout * 1000);
        if (d == null) {
            LOG.info(" [RabbitReader] timout, exiting reader!");
            return false;
        }
        nextDelivery = d.getBody();
        deliveryTag = d.getEnvelope().getDeliveryTag();
        return true;

    } catch (InterruptedException ie) {
        LOG.info(" [RabbitReader] timout2, exiting reader!");
        return false;
    } catch (Exception e) {
        throw new CollectionException(e);
    }
}

RabbitSupplier( RabbitConnection connection, String queueName, String bindTopic, String bindRoutingKey,
                Consumer<Delivery> optionalConsumer,
                boolean durable,
                Map<String, Object> queueProperties )
{
    this.connection = Objects.requireNonNull( connection );
    this.queueName = Objects.requireNonNull( queueName );
    this.bindTopic = bindTopic;
    this.bindRoutingKey = bindRoutingKey;
    this.durable = durable;

    this.optionalConsumer = optionalConsumer;
    supplier = optionalConsumer == null ? new BlockingSupplierConsumer<>( 1 ) : null;

    if( durable ) {
        this.queueProperties = queueProperties == null ? new HashMap<>() : queueProperties;

        this.queueProperties.putIfAbsent( "x-message-ttl", DEFAULT_TTL );
    }
    else {
        // queue properties are optiona, so null is valid. If empty then replace with null
        this.queueProperties = (queueProperties == null || queueProperties.isEmpty()) ? null : queueProperties;
    }

    // Enable parking
    if( this.queueProperties != null && this.queueProperties.containsKey( "internal-park-queue" ) ) {
        parked = true;
        parkDelay = (Long) this.queueProperties.remove( "internal-park-queue" );
        parkQueueName = queueName + "$park";
    }
    else {
        parked = false;
        parkDelay = 0L;
        parkQueueName = null;
    }
}

private void processNextMessage()
        throws IOException,
               InterruptedException
{
    final QueueingConsumer.Delivery delivery = getQueueingConsumer().
            nextDelivery();

    try {
        if( optionalConsumer == null ) {
            supplier.accept( delivery.getBody() );
        }
        else {
            optionalConsumer.accept( delivery );
        }

        // ack as we have handled it safely
        channel.basicAck( delivery.getEnvelope().
                getDeliveryTag(), false );

    }
    catch( Exception ex ) {
        // nack and don't requeue
        getChannel().
                basicNack( delivery.getEnvelope().
                        getDeliveryTag(), false, false );
    }
}

private void processDelivery(Delivery delivery) throws Exception {
    BasicProperties props = delivery.getProperties();
    BasicProperties replyProps = new BasicProperties.Builder().correlationId(props.getCorrelationId()).build();
    DataInputStream data = new DataInputStream(new InflaterInputStream(new ByteArrayInputStream(delivery.getBody())));
    RenderMode mode = RenderMode.values()[data.readUnsignedByte()];
    int width = data.readUnsignedShort();
    int height = data.readUnsignedShort();
    GameProfile profile = Profiles.readGameProfile(data);
    Map<String, String[]> params = Maps.newHashMap();
    int len = data.readUnsignedShort();
    for (int i = 0; i < len; i++) {
        String key = data.readUTF();
        String[] val = new String[data.readUnsignedByte()];
        for (int v = 0; v < val.length; v++) {
            val[v] = data.readUTF();
        }
        params.put(key, val);
    }
    byte[] skinData = new byte[data.readInt()];
    data.readFully(skinData);
    BufferedImage skin = new PngImage().read(new ByteArrayInputStream(skinData), false);
    Visage.log.info("Received a job to render a "+width+"x"+height+" "+mode.name().toLowerCase()+" for "+(profile == null ? "null" : profile.getName()));

    RenderConfiguration conf = new RenderConfiguration(Type.fromMode(mode), Profiles.isSlim(profile), mode.isTall(), Profiles.isFlipped(profile));

    glClearColor(0, 0, 0, 0);
    glClear(GL_COLOR_BUFFER_BIT | GL_DEPTH_BUFFER_BIT);
    byte[] pngBys = draw(conf, width, height, profile, skin, params);
    if (Visage.trace) Visage.log.finest("Got png bytes");
    parent.channel.basicPublish("", props.getReplyTo(), replyProps, buildResponse(0, pngBys));
    if (Visage.trace) Visage.log.finest("Published response");
    parent.channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    if (Visage.trace) Visage.log.finest("Ack'd message");
}

public void nextTuple() {
    try {
        Delivery d = consumer.nextDelivery(0);
        if (d != null) {
            long tag = d.getEnvelope().getDeliveryTag();
            int an_id = r.nextInt();
            out_id.put(an_id, tag);
            collector.emit(new Values(d.getBody()), an_id);
        }
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException(e);
    }
}

@Override
public void run() {

    try {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, "fanout", durable);

        DeclareOk ok = channel.queueDeclare(queue_name, durable, false, false, null);
        String queueName = ok.getQueue();

        channel.queueBind(queueName, EXCHANGE_NAME, "");

        System.out.println(" [" + queue_name + "] Waiting for messages. To exit press CTRL+C");

        channel.basicQos(1);    // 消息分发处理
        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(queueName, false, consumer);

        while (true) {
            Thread.sleep(2000);
            Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" [" + queue_name + "] Received '" + message + "'");
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    } catch (Exception e) {
        e.printStackTrace();
    }
}

@Override
public final Reply receiveMessage(Request message) throws IOException, ProActiveException {
    RpcReusableChannel channel = getRpcReusableChannel();
    try {
        if (logger.isDebugEnabled()) {
            logger.debug(String.format("AMQP RO sending %s to %s, on exchange %s, queue %s", message
                    .getMethodName(), remoteObjectURL, rpcExchangeName, queueName));
        }

        String replyQueue = channel.getReplyQueue();
        byte[] messageBody = ObjectToByteConverter.ProActiveObjectStream.convert(message);
        channel.getChannel().basicPublish(rpcExchangeName, queueName,
                new BasicProperties.Builder().replyTo(replyQueue).build(), messageBody);

        while (true) {
            Delivery delivery = channel.getReplyQueueConsumer().nextDelivery(replyTimeout);
            if (delivery != null) {
                Reply reply = (Reply) ByteToObjectConverter.ProActiveObjectStream.convert(delivery
                        .getBody());
                if (logger.isDebugEnabled()) {
                    logger.debug(String.format(
                            "AMQP RO received response of message %s to %s, on exchange %s, queue %s",
                            message.getMethodName(), remoteObjectURL, rpcExchangeName, queueName));
                }
                channel.returnChannel();
                return reply;
            } else {
                // if didn't receive reply after timeout expired then check that remote object server still exists
                checkTargetObjectExists();
            }
        }
    } catch (Throwable e) {
        channel.close();

        throw new IOException(String.format("AMQP cannot send %s to %s, on exchange %s, queue %s",
                message.getMethodName(), remoteObjectURL, rpcExchangeName, queueName), e);
    }
}

@Override
public void startup() {
    //Delivery Actor
    ActorRef delivery = system.actorOf(Props.create(DeliveryActor.class), "delivery");
    system.eventStream().subscribe(delivery, Delivery.class);

    //Listening Actor
    system.actorOf(Props.create(ListeningActor.class), "listening");
}

@Override
protected Runnable buildMsgProcessingTask(Delivery delivery) {
    return new MsgProcessingTask(delivery, this);
}

public MsgProcessingTask(Delivery delivery, ResponseHandler responseHandler) {
    this.delivery = delivery;
    this.responseHandler = responseHandler;
}

public DeliveryProcessing(StorageService storage, Delivery delivery, RabbitQueue queue) {
    this.storage = storage;
    this.delivery = delivery;
    this.queue = queue;
}

public void process(Delivery delivery) throws IOException {
    toProcess.addLast(delivery);
}

/**
 * 接口方法实现消息的处理
 * @param delivery
 */
void processMessage(Delivery delivery) throws Exception;