:Notas de versión del SDK para Java

Notas de uso

• Solo puede actualizar su SDK de cliente TCP para Java a la versión V2.x.x.Final en las siguientes regiones: China (Hangzhou), China (Qingdao), China (Beijing), China (Zhangjiakou), China (Hohhot), China (Shenzhen), China (Chengdu), China (Hong Kong), Alemania (Frankfurt) e Indonesia (Jakarta). No actualice su SDK de cliente TCP para Java a la versión V2.x.x.Final en otras regiones; de lo contrario, no podrá usar el SDK para conectarse a ApsaraMQ for RocketMQ.

• Puede usar el SDK de cliente TCP para Java 2.x.x.Final para acceder a ApsaraMQ for RocketMQ únicamente dentro de redes privadas virtuales (VPC).

  Si utiliza una instancia existente de ApsaraMQ for RocketMQ y accede a ella mediante la red clásica, no actualice el SDK de cliente TCP para Java a la versión 2.x.x.Final. De hacerlo, perderá el acceso a la instancia de ApsaraMQ for RocketMQ.

• El SDK de cliente TCP para Java 2.x.x.Final solo permite acceder a instancias que contienen espacios de nombres. Si la instancia que utiliza no tiene un espacio de nombres, no actualice el SDK de cliente TCP para Java a la versión 2.x.x.Final.

  De forma predeterminada, todas las instancias de ApsaraMQ for RocketMQ 5.x incluyen espacios de nombres. Si usa una instancia de ApsaraMQ for RocketMQ 4.x, puede comprobar si la instancia cuenta con un espacio de nombres en la sección Basic Information de la página Instance Details en ApsaraMQ for RocketMQ console.

Información de versiones

Requisitos del entorno

Para utilizar el SDK para Java V2.x.x, asegúrese de contar con un Java Development Kit (JDK) compatible con Java 8 o superior.

Cambios en funciones de la versión 2.0.5

Función optimizada

Se admite el registro asíncrono.

Problemas corregidos

• No era posible especificar el tiempo de espera para el consumo por lotes.

• Se corrigieron vulnerabilidades de seguridad específicas.

Cambios en funciones de la versión 2.0.3

Fixed issue

En versiones recientes del JDK, no era posible aumentar el número de hilos en un pool de hilos a un valor superior a 32.

Cambios en funciones de la versión 2.0.2

Fixed issue

Podía producirse un interbloqueo durante el envío de mensajes.

Cambios en funciones de la versión 2.0.1

Message trace

Los resultados de consulta de rastreo ahora devuelven más datos.

Cambios en funciones de la versión 2.0.0

Ordered message

El valor predeterminado del parámetro MaxReconsumeTimes cambió de Integer.MAX a 16. Este parámetro define el número máximo de reintentos para mensajes ordenados. Si un consumidor sigue sin poder procesar un mensaje tras alcanzar este límite, el mensaje se envía a la cola de mensajes muertos. Puede modificar el valor del parámetro MaxReconsumeTimes para ajustar el número máximo de reintentos de los mensajes ordenados.

Transactional message

Cuando un productor envía mensajes, se genera una excepción si la clase LocalTransactionExecutor es nula. En versiones anteriores, este error no generaba ninguna excepción.

Broadcasting consumption

En el modo de consumo por difusión, puede invocar la operación offsetStore para especificar el desplazamiento del consumidor desde el cual inicia el consumo. Si no define este valor, el consumo comienza desde el desplazamiento más reciente, tal como ocurría en versiones anteriores.

public class BroadcastingConsumerExample {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();

        properties.put(PropertyKeyConst.GROUP_ID, "MyGroupId");
        properties.put(PropertyKeyConst.AccessKey, "MyAccessKey");
        properties.put(PropertyKeyConst.SecretKey, "MySecretKey");
        // The TCP endpoint. You can obtain the endpoint in the TCP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console. 
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXXX");
        // You can obtain the consumer offset by calling the offsetStore operations only in broadcasting consumption mode. 
        properties.put(PropertyKeyConst.MessageModel, MessageModel.BROADCASTING);

        Consumer consumer = ONSFactory.createConsumer(properties);
        // The frequency of calls for the AbstractOffsetStore method. The value 1 specifies that the system calls the AbstractOffsetStore method every second to set a persistent offset. 
        OffsetStore offsetStore = new AbstractOffsetStore(1) {
            @Override
            public Map<TopicPartition, Long> loadOffset() {
            // The logic that is used to obtain the offset from an external storage system. 
            }

            @Override
            public void persistOffset(Map<TopicPartition, Long> offsetTable) {
            // The logic that is used to persist the offset to an external storage system. 
            }
        };
        offsetStore.start();
        consumer.setOffsetStore(offsetStore);

        consumer.subscribe("testBroadcastingTopic", "TagA", new MessageListener() {
            @Override
            public Action consume(Message message, ConsumeContext context) {
                // The logic that is used to consume messages. 
                return Action.CommitMessage;
            }
        });

        consumer.start();
        Thread.sleep(100000);
        consumer.shutdown();
        offsetStore.shutdown();
    }
}

Consumption in Push mode

• No se admite el consumo por lotes de mensajes normales.

• Si el número especificado de hilos de consumo no se encuentra dentro del rango válido de 1 a 1,000, el sistema genera una excepción al crear el consumidor.

• Se agregó la función de limitación de consumo. Esta opción permite restringir la tasa de procesamiento de mensajes para evitar excepciones en la aplicación provocadas por aumentos repentinos de tráfico en los clientes consumidores. El siguiente código muestra cómo configurar una tasa personalizada de consumo de mensajes.

  Nota

  La limitación de consumo no aplica a los reintentos de mensajes ordenados.

  public class RateLimitConsumerExample {
      public static void main(String[] args) throws InterruptedException {
          Properties properties = new Properties();
  
          properties.put(PropertyKeyConst.GROUP_ID, "MyGroupId");
          properties.put(PropertyKeyConst.AccessKey, "MyAccessKey");
          properties.put(PropertyKeyConst.SecretKey, "MySecretKey");
          // The TCP endpoint. You can obtain the endpoint in the TCP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console. 
          properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXX");
  
          Consumer consumer = ONSFactory.createConsumer(properties);
  
          // Specify the message consumption rate in testTopicA. In this example, the consumer client can consume only 10 messages per second. 
          consumer.rateLimit("testTopicA", 10);
          consumer.subscribe("testTopic", "TagA", new MessageListener() {
              @Override
              public Action consume(Message message, ConsumeContext context) {
                  // The logic that is used to consume messages. 
                  return Action.CommitMessage;
              }
          });
  
          consumer.start();
          Thread.sleep(100000);
          consumer.shutdown();
      }
  }

Consumption in Pull mode

No se admite el consumo en modo Pull.

Log configuration

• La ruta predeterminada de registros cambió de ~/logs/ons.log a ~/logs/ons/ons-client.log.

• Los niveles de registro ahora son consistentes con los niveles compatibles con el framework logback. Se agregaron los niveles OFF, TRACE y ALL. Las versiones anteriores solo admitían los niveles ERROR, WARN, INFO y DEBUG.

• Ahora se admiten variables de entorno para agregar parámetros relacionados con los registros. Las versiones anteriores solo admitían -D.

Client creation

Si se especifica un endpoint no válido, se genera una excepción cuando el sistema intenta crear un productor o un consumidor.

Message trace

Problemas corregidos en la versión 2.0.0

Se corrigió el siguiente problema: al invocar la operación updateCredential varias veces en un periodo corto, la autorización entre cuentas para roles RAM mediante Security Token Service (STS) fallaba porque no se garantizaba la atomicidad al actualizar el AccessKey ID, el AccessKey secret y el STS token.

Para consultar el código de ejemplo del SDK sobre la autorización STS, consulte Step 2: Access resources across Alibaba Cloud accounts.