Traitement des messages à l'aide d'un fournisseur Kafka

Avec un fournisseur de messagerie Kafka , vous pouvez traiter les messages sortants de manière séquentielle et les messages entrants de manière séquentielle ou continue. En général, le traitement des messages Kafka est similaire au traitement des messages JMS.

Traitement des messages Kafka

Le traitement des messages Kafka effectue le suivi des messages traités en conservant une valeur de décalage qui représente l'ordre séquentiel dans lequel les messages sont reçus par les rubriques Kafka . La valeur de décalage indique le prochain message à traiter. Une fois qu'un message a été traité dans Kafka, le décalage est désormais supérieur à ce message traité et le message n'est plus jamais traité.

Le message traité n'est pas supprimé de la file d'attente. Il subsiste dans celle-ci jusqu'à ce que le délai d'expiration configuré pour lui soit écoulé.

Maximo® Manage fournit une application qui vous aide à retraiter les messages comportant des erreurs. Voir plus d'informations sur le traitement des erreurs.

Limites de taille des messages

Vous configurez une limite de taille de message sur le serveur Kafka . Ce faisant, tenez compte de la taille de vos transactions, y compris de celle des pièces jointes intégrées. Une trop grande taille limite peut avoir un impact négatif sur les performances de lecture et d'écriture des messages dans la file d'attente.

Dans Maximo Manage, vous pouvez ajouter la propriété système mxe.kafka.messagesize pour la taille maximale des messages pouvant être traités. Configurez la propriété pour qu'elle corresponde à la taille du message, en octets, que vous définissez sur le serveur Kafka . La valeur par défaut est 10 Mo.

Kafka dépassements du délai d'écriture

Dans Maximo Manage, les actions d'écriture Kafka attendent un accusé de réception des courtiers Kafka . Vous pouvez contrôler le délai d'écriture du producteur Kafka à l'aide de la propriété mxe.kafka.waittimeforack . L'unité de mesure est en millisecondes.

Traitement des messages séquentiels sortants

L'infrastructure d'intégration utilise des rubriques Kafka en tant que files d'attente de messages séquentielles sortantes. Lorsqu'un message est envoyé à partir de l'infrastructure d'intégration, il est acheminé vers votre fournisseur Kafka , où il est hébergé sous forme de message JSON dans une rubrique prédéfinie. Bien que les canaux de publication puissent envoyer des messages au format XML ou JSON, le produit les encapsule en tant que messages au format JSON et les écrit dans la rubrique Kafka . Un message sortant utilise la structure illustrée dans l'exemple de code suivant :
{
      “interfacetype”:”MAXIMO”,
      “INTERFACE”:”MXPRInterface”,
      “payload”:”<xml/json message>”,
      “SENDER”:”MX”,
      “destination”:”<external system name>”,
      “destjndiname”:”<kafka topic name>”,
      “compressed”:”1”,
      “MEAMessageID”:”<providername>~<topic>~<partition>~<offset>”,
      “mimetype”:”application/xml”
}
La propriété payload contient la charge utile envoyée au point d'extrémité. L'intégralité du message Kafka est compressée au moment du stockage dans les partitions Kafka . Dans les partitions Kafka , les messages sont stockés dans un ordre d'horodatage strict. La position du message dans la partition est donnée par un numéro d'ID séquentiel, appelé décalage (offset) du message. Les rubriques Kafka peuvent être partitionnées, mais celles qui sont traitées séquentiellement doivent se trouver dans une même partition.

Une fois le message écrit dans la rubrique Kafka , il est traité par la tâche périodique Kafka associée que vous configurez dans l'application Configuration des tâches périodiques pour chaque rubrique. Le traitement se produit en fonction du système externe et du point d'extrémité configurés pour le canal.

Traitement séquentiel des messages entrants

Le traitement des messages entrants peut être séquentiel ou continu. Le traitement séquentiel des messages entrants est similaire au traitement séquentiel des messages sortants et nécessite les mêmes types de configuration et notamment une instance de la tâche périodique Kafka.

La structure d'un message entrant séquentiel est illustrée dans l'exemple suivant :
{
     “interfacetype”: "MAXIMO",
     “SENDER”: "testkafka",
     “destination”: "testkafka",
     “USER”: "wilson",
     “MEAMessageID”: "….",
     “INTERFACE”: "MXASSETInterface",
     “payload”: "{"assetnum":"AZ163","siteid":"BEDFORD"}",
     “mimetype”: "application/json",
    “destjndiname”: "anamitratestcont"
}

Configurez chaque file d'attente séquentielle avec une unique partition et pas plus d'une instance de tâche périodique pour consommer les messages de cette file.

Voir plus d'informations sur le traitement des erreurs.

Traitement et mise à l'échelle des files d'attente continues sortantes et entrantes

D'une manière générale, et contrairement au traitement des files séquentielles, celui des files continues ne s'interrompt pas lorsqu'un message présente une erreur. Les messages comportant des erreurs sont écrits dans la table d'erreurs et traités à partir de la table d'erreurs par l'instance de tâche périodique KAFKAERROR pour la file d'attente correspondante.

Une seule file d'attente Kafka peut être mise à l'échelle en tant que file d'attente à unités d'exécution multiplesou en tant que file d'attente comportant plusieurs destinataires de messages, en définissant d'autres partitions pour la file d'attente. Chaque partition représente alors un consommateur séparé, et chacune nécessite sa propre instance de tâche périodique pour prendre en charge sa consommation de messages. Pour augmenter la capacité de traitement, configurez plusieurs partitions. Le nombre d'instances de tâche périodique doit être identique au nombre de partitions.

La file d'attente continue sortante écrit des messages sortants pour les interfaces si elles sont sélectionnées dans l'application Systèmes externes de l'application Maximo Manage . Plusieurs systèmes externes peuvent utiliser la même file d'attente continue sortante.

Le traitement des messages entrants continus avec Kafka est similaire au traitement des messages JMS, mais il existe des différences dans la mise à l'échelle et le traitement des erreurs.

Voir plus d'informations sur le traitement des erreurs des messages dans les files d'attente continues et la configuration d'un délai de redistribution.