import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
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 + "'");
		}
	}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
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;
		}
	}
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
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);
	}
	
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
@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);
    }
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
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;
    }
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
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 );
    }
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
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");
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
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);
	}
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
@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();
	}
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
@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);
    }
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
@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");
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
@Override
protected Runnable buildMsgProcessingTask(Delivery delivery) {
    return new MsgProcessingTask(delivery, this);
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
public MsgProcessingTask(Delivery delivery, ResponseHandler responseHandler) {
    this.delivery = delivery;
    this.responseHandler = responseHandler;
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
public DeliveryProcessing(StorageService storage, Delivery delivery, RabbitQueue queue) {
    this.storage = storage;
    this.delivery = delivery;
    this.queue = queue;
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
public void process(Delivery delivery) throws IOException {
	toProcess.addLast(delivery);
}
 

import com.rabbitmq.client.QueueingConsumer.Delivery; //导入依赖的package包/类
/**
 * 接口方法实现消息的处理
 * @param delivery
 */
void processMessage(Delivery delivery) throws Exception;