Analyse des instances de file d'attente de messages Kafka distribuée Java

Introduction

Apache Kafka est un système de messagerie de publication-abonnement distribué. La définition de kafka sur le site officiel de kafka est : un système de messagerie de publication-abonnement distribué. Il a été initialement développé par LinkedIn, qui a contribué à la Fondation Apache en 2010 et est devenu l'un des principaux projets open source. Kafka est un service de journal de validation rapide, évolutif et intrinsèquement distribué, partitionné et réplicable.

Remarque : Kafka ne suit pas la spécification JMS (), il fournit uniquement des méthodes de communication de publication et d'abonnement.

Noms associés au noyau Kafka

1. Courtier : nœud Kafka, un nœud Kafka est un courtier, plusieurs courtiers peuvent former un cluster Kafka

2. Sujet : un type de message, le répertoire où le message est stocké est le sujet , tels que les journaux d'affichage de page, les journaux de clics, etc. peuvent exister sous forme de sujets. Le cluster Kafka peut être responsable de la distribution de plusieurs sujets en même temps

3. massage : l'objet de diffusion le plus basique de Kafka.

4. Partition : le regroupement physique de sujets. Un sujet peut être divisé en plusieurs partitions, et chaque partition est une file d'attente ordonnée. Le partitionnement est implémenté dans Kafka et un courtier représente une région.

5. Segment : La partition est physiquement composée de plusieurs segments, chaque segment stocke les informations sur les messages

6. Producteur : Producteur, produit des messages et les envoie au sujet

7. Consommateur : Consommateur, s'abonne au sujet et consomme des messages, consommateur consomme comme un fil de discussion

8. Groupe de consommateurs : groupe de consommateurs, un groupe de consommateurs contient plusieurs consommateurs

9. Offset : décalage, compris comme la position d'index du message dans la partition de messages

topic et La différence entre les files d'attente :

La file d'attente est une structure de données qui suit le principe du premier entré, premier sorti

Installation du cluster Kafka

• Installer l'environnement jdk1.8 sur chaque serveur

• Installer l'environnement de cluster Zookeeper

• Installer kafka cluster Environment

• Exécution du test d'environnement

L'installation de l'environnement jdk et de zookeeper ne sera pas détaillée ici.

Pourquoi kafka dépend-il de zookeeper : Kafka stockera les informations mq sur zookeeper Afin de faciliter l'expansion de l'ensemble du cluster, la notification d'événement de zookeeper est utilisée pour se détecter.

Étapes d'installation du cluster Kafka :

1. Téléchargez le package compressé de Kafka

2 Décompressez le package d'installation

tar -zxvf kafka_2.11-1.0.0.tgz

3. .properties

Contenu de la modification du fichier de configuration :

• Adresse de connexion zookeeper : zookeeper.connect=192.168.1.19:2181zookeeper.connect=192.168.1.19:2181

• 监听的ip，修改为本机的iplisteners=PLAINTEXT://192.168.1.19:9092

• kafka的brokerid，每台broker的id都不一样broker.id=0

4、依次启动kafka

./kafka-server-start.sh -daemon config/server.properties

kafka使用

kafka文件存储

topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是Producer生成的数据。Producer生成的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制，将每个partition分为多个segment，每个segment包括：“.index”文件、“.log”文件和.timeindex等文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号。

例如：执行命令新建一个主题，分三个区存放放在三个broker中：

./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic kaico

L'adresse IP d'écoute est remplacée par l'adresse IP localeauditeurs =PLAINTEXT://192.168.1.19:9092

brokerid de Kafka, l'identifiant de chaque courtier est différent broker.id=0

• 4 Démarrez kafka dans l'ordre.
• ./kafka-server-start.sh -daemon config/server.properties
  kafka utilise
• le stockage de fichiers Kafka
  le sujet est un concept logique et la partition est physique Basé sur le concept ci-dessus, chaque partition correspond à un fichier journal, et le fichier journal stocke les données générées par le Producteur. Les données générées par le producteur seront continuellement ajoutées à la fin du fichier journal. Afin d'éviter que le fichier journal ne soit trop volumineux et ne provoque une inefficacité dans le positionnement des données, Kafka adopte un mécanisme de partitionnement et d'indexation pour diviser chaque partition en plusieurs segments. . Chaque segment comprend : des fichiers ".index", des fichiers ".log" et des fichiers .timeindex. Ces fichiers se trouvent dans un dossier et la règle de dénomination du dossier est la suivante : nom du sujet + numéro de série de la partition.
• Par exemple : exécutez la commande pour créer un nouveau sujet, divisé en trois zones et stocké dans trois courtiers :
  ./kafka-topics.sh --create --zookeeper localhost:2181 --replication -facteur 1 - -partitions 3 --topic kaico

Une partition est divisée en plusieurs segments

.fichier journal

🎜🎜.index fichier d'index de décalage🎜 🎜🎜 🎜.fichier d'index d'horodatage .timeindex🎜🎜🎜🎜Autres fichiers (partition.metadata, leader-epoch-checkpoint)🎜🎜🎜🎜Springboot intègre kafka🎜🎜dépendance maven🎜

 <dependencies>
        <!-- springBoot集成kafka -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <!-- SpringBoot整合Web组件 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>

🎜configuration yml🎜

# kafka
spring:
  kafka:
    # kafka服务器地址(可以多个)
#    bootstrap-servers: 192.168.212.164:9092,192.168.212.167:9092,192.168.212.168:9092
    bootstrap-servers: www.kaicostudy.com:9092,www.kaicostudy.com:9093,www.kaicostudy.com:9094
    consumer:
      # 指定一个默认的组名
      group-id: kafkaGroup1
      # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
      auto-offset-reset: earliest
      # key/value的反序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      # key/value的序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 批量抓取
      batch-size: 65536
      # 缓存容量
      buffer-memory: 524288
      # 服务器地址
      bootstrap-servers: www.kaicostudy.com:9092,www.kaicostudy.com:9093,www.kaicostudy.com:9094

🎜producteur 🎜

@RestController
public class KafkaController {
	/**
	 * 注入kafkaTemplate
	 */
	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;
	/**
	 * 发送消息的方法
	 *
	 * @param key
	 *            推送数据的key
	 * @param data
	 *            推送数据的data
	 */
	private void send(String key, String data) {
		// topic 名称 key   data 消息数据
		kafkaTemplate.send("kaico", key, data);
	}
	// test 主题 1 my_test 3
	@RequestMapping("/kafka")
	public String testKafka() {
		int iMax = 6;
		for (int i = 1; i < iMax; i++) {
			send("key" + i, "data" + i);
		}
		return "success";
	}
}

🎜consommateur 🎜

@Component
public class TopicKaicoConsumer {
    /**
     * 消费者使用日志打印消息
     */
    @KafkaListener(topics = "kaico") //监听的主题
    public void receive(ConsumerRecord<?, ?> consumer) {
        System.out.println("topic名称:" + consumer.topic() + ",key:" +
                consumer.key() + "," +
                "分区位置:" + consumer.partition()
                + ", 下标" + consumer.offset());
        //输出key对应的value的值
        System.out.println(consumer.value());
    }
}

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!