Java异步消息操作RocketMQ的第八部分主要是实现消息的消费者，具体的代码实现步骤如下：
1. 首先创建一个消费者对象，代码如下：
“`
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
“`
其中，"consumer_group"是消费者组的名称，可以自定义。
2. 设置消费者的相关属性，包括nameserver地址、订阅的topic和tag等，代码如下：
“`
consumer.setNamesrvAddr("localhost:9876");
consumer.subscribe("test_topic", "*");
“`
其中，"localhost:9876"是nameserver的地址，"test_topic"是订阅的topic名称，"*"表示订阅所有的tag。
3. 注册消息监听器，代码如下：
“`
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
System.out.println("Received message: " + new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
“`
其中，MessageListenerConcurrently是RocketMQ提供的消息监听器接口，用于处理并发消费的情况。在consumeMessage方法中，可以处理接收到的消息，这里只是简单地打印出消息内容。
4. 启动消费者，代码如下：
“`
consumer.start();
“`
5. 最后，在程序结束时关闭消费者，代码如下：
“`
consumer.shutdown();
“`
以上就是Java异步消息操作RocketMQ的第八部分的代码实现步骤。需要注意的是，在实际应用中，还需要处理消费者的异常情况、消费者的负载均衡等问题，以保证消息的可靠消费。