Shulou(Shulou.com)06/01 Report--

This article mainly introduces the relevant knowledge of "what are the concepts of Kubernetes message queue". The editor shows you the operation process through an actual case, and the operation method is simple, fast and practical. I hope this article "what are the concepts of Kubernetes message queue" can help you solve the problem.

Quality of service

Service semantics

Quality of service can generally be divided into three levels, which are described below with different semantics.

At most once

At most once, the message may be lost, but it will never be transmitted repeatedly. Producer: completely depends on the transmission reliability of the underlying TCP/IP, without special treatment, the so-called "send and forget". Set acks=0 in kafka. Consumers: save the consumption progress before processing the message. Set the consumer autocommit offset in kafka and set a short commit interval.

At least once

At least once, the message will never be lost, but it may be repeated. Producer: make a guarantee against the loss of messages. Set acks=1 or all in kafka and set retries > 0. Consumers: first process the message, and then save the consumption progress. Set the consumer autocommit offset in kafka and set a long commit interval, or turn off the autocommit offset directly, and manually invoke the offset submission of synchronous mode after processing the message.

Exactly once

Exactly once, each message is sure to be transmitted once and only once. This level is not guaranteed by the message queue itself, and may depend on external components. Producer: make a guarantee against the loss of messages. Set acks=1 or all in kafka and set retries > 0. In mosquito, single semantics is realized by four-step handshake and DUP, MessageID and other logos. Consumers: to ensure that messages are not duplicated, there are a variety of solutions, such as: introducing a two-phase commit protocol in the two operations of saving consumption progress and processing messages; making messages idempotent; keep consumption processing and progress saving in a single transaction to ensure atomicity. Disable autocommit offsets in kafka and set custom rebalance listeners to read or store offsets from external components when partitions change, so as to ensure that they or other consumers can read the latest offsets when changing partitions to avoid repetition. In short, it is the combination of ConsumerRebalanceListener, seek and an external system (such as a database that supports transactions) to achieve single semantics. In addition, kafka also provides GUID for users to remove weight on their own. Kafka 0.11 supports EOS:1 with 3 major changes. Idempotent producer;2. Support transactions; 3. Supports streaming of EOS (EOS that guarantees read-process-write full links). These three levels of reliability will increase in turn, but latency and bandwidth usage will also increase, so in practice, trade-offs should be made according to the type of business.

Reliability.

The above three semantics need not only the cooperation of producers and consumers, but also the reliability of broker itself. Reliability means that as long as broker sends an acknowledgement to producer, you must ensure that the message can be obtained by consumer.

A topic in a kafka has multiple partition, and each partition has multiple replica. One of the replica must synchronize the leader before returning the successful replica set. The replica in the OSR tries his best to synchronize the leader, and the data version may lag behind. In the process of kafka work, if the synchronization speed of a replica is slower than the threshold specified by replica.lag.time.max.ms, it will be kicked out of ISR and saved in OSR, and can be returned to ISR if the subsequent speed is restored. You can configure min.insync.replicas to specify the minimum number of replica in ISR, which defaults to 1. LEO is the offset of the latest data of the partition, and when the data is written to the leader, the LEO immediately executes the latest data, which is equivalent to the latest data identification bit. HW means that when the written data is synchronized to all replicas in ISR, the data is considered to have been submitted, and the HW is updated to this location, so that the data before HW can be accessed by consumers, ensuring that the data without synchronization will not be accessed by consumers, which is equivalent to all replicas synchronizing data identification bits.

All replica of each partition requires a leader election (dependent on ZooKeeper). After the leader goes down, you can only select a new leader from the ISR list. No matter which copy of the ISR is selected as the new leader, it knows the data before the HW, which ensures that after switching the leader, consumers can continue to see the previously submitted data of the HW. When all the replica in the ISR is down, the partition is not available. You can set unclean.leader.election.enable=true. This option allows the kafka to select any live replica to become a leader and then continue to work. This replica may not be in the ISR, which may result in data loss. Therefore, it is necessary to make a tradeoff between availability and reliability in practical use.

Kafka recommends that reliable data storage does not rely on data forced flushing (which affects overall performance), but on replica.

Sequential consumption

Sequential consumption means that the order in which consumers process messages is consistent with the order in which producers put messages. The main scenario that may break the order is that the producer sends two messages AB, and then A fails to redeliver, resulting in the consumer getting the message BA.

The ordering of messages within the partition can be guaranteed in kafka by setting max.in.flight.requests.per.connection=1, that is, producers will not send message B without the confirmation of message A by broker. This ensures that the messages stored by broker are orderly, and the messages requested by natural consumers are also orderly. But we can clearly feel that this will reduce throughput because messages can no longer be delivered in parallel, block waits, and fail to harness the power of batch. If you want the whole topic to be ordered, you can only have one topic, one partition, and one consumer group has only one consumer. This goes against the original intention of kafka's high throughput.

Repeated consumption

Repeated consumption means that a message is repeatedly consumed by consumers. This problem also needs to be solved in the third semantics above.

General messaging systems such as kafka or similar rocketmq cannot and do not encourage solutions within the system, but cooperate with third-party components to let users solve them. The reason is that the cost of solving the problem does not match the value gained after solving the problem, so it simply does not solve it, just like the operating system treats deadlocks and adopts an "ostrich policy". But kafka 0.11 still deals with this problem, see the release notes, maintainers are trying to make users impeccable [laugh at cry].

Performance

There are many ways to measure the performance of a messaging system, the most common of which are the following metrics.

Number of connections

Refers to the total number of connections to how many producers or consumers the system can support at the same time. The number of connections is directly related to the network IO model adopted by broker. The common models are: single thread, connection per thread, Reactor, Proactor and so on. A single thread can only handle one connection at a time, and each thread of connection is limited by the number of threads in server. Reactor is the mainstream high-performance network IO model at present. Proactor is not popular because the operating system does not support true asynchronism.

Kafka's broker uses a Reactor model similar to Netty: 1 (1 acceptor thread) + N (N process threads) + M (M work threads). Acceptor is responsible for listening for new connection requests, registering OPACCEPT events, and handing over the new connection to a processor thread for processing in a RoundRobin manner. Each Processor has a NIO selector, registers OPREAD and OPWRITE events to the SocketChannel assigned to the Acceptor, and reads and writes to the socket. N is determined by num.networker.threads. Worker is responsible for specific business logic such as reading requests from requestQueue, storing data to disk, putting responses into responseQueue, and so on. The size of M is determined by num.io.threads.

The Reactor model is generally based on IO multiplexing (such as select,epoll) and is non-blocking, so a small number of threads can handle a large number of connections. If a large number of connections are idle, then the efficiency of Reactor using epoll is leveraged, and if a large number of connections are active, it is best to replace epoll with select or poll without the support of Proactor. The way to do this is to change the EPollSelectorProvider under the sun.nio.ch package to PollSelectorProvider by Djava.nio.channels.spi.SelectorProvider.

QPS

Refers to the number of requests that the system can process per second. QPS usually reflects the size of throughput (the term is broad and can be expressed in units such as TPS/QPS, PV, UV, number of businesses per hour, and so on).

Because batch can be used in kafka (and can also be compressed), a large number of requests can be processed per second (because of reduced parsing, network round-tripping, disk IO, and so on). On the other hand, kafka has multiple partition per topic, so multiple producers and consumers can be served in parallel (not concurrently) under the same topic, which also improves throughput.

Average response time

The average response time is the waiting time for each request to get a response.

Where are the bottlenecks in processing requests (that is, the factors that most affect response time) most likely to occur in kafka? The Internet? It is possible, but generally speaking, this factor is beyond the control of kafka. Kafka can encode and compress messages and submit them in batches to reduce bandwidth consumption. It is very possible, so kafka uses OS's pagecache separately and writes sequentially to the disk, which greatly improves the write speed of the disk. At the same time, kafka also uses zero-copy technology, which omits the copying process from kernel buffer to user buffer in the common copy process: disk- > read buffer- > app buffer- > socket buffer- > NIC buffer, which speeds up the processing speed. In addition, there is file segmentation technology, each partition is divided into multiple segment, avoiding large file operations while improving the degree of parallelism. CPU? Unlikely, because the use of message queues does not involve a lot of computation, and common consumption includes thread switching, codec, compression and decompression, memory copying, and so on, which are generally not bottlenecks in big data's processing.

Concurrent number

Is the number of requests that the system can process at the same time. Generally speaking, QPS = number of concurrency / average response time or number of concurrency = average response time of QPS*.

Generally speaking, this parameter can only be estimated or calculated, but can not be measured directly. As the name implies, the better the machine performance, the higher the number of concurrency. In addition, using multithreading technology and improving the parallelism of code, optimizing the IO model, and reducing memory allocation and release can all increase the number of concurrency.

Expansibility

The extensibility of the messaging system means that it is easier to add new members to the system components.

The cornerstone of extensibility in kafka is the partition mechanism adopted by topic. First, Kafka solves the data skew problem by allowing Partition to move between Broker in cluster. Second, support custom Partition algorithms, for example, you can route all messages from the same Key to the same Partition (to get the order). Third, all replica of partition are managed through ZooKeeper for cluster management, and replicas can be dynamically added or decreased. Fourth, partition also supports dynamic increase and decrease.

For producer, there is no extension problem, as long as broker is enough for you to connect. For consumer, the consumer in a consumer group can be increased or decreased, but it is best not to exceed the number of partition in a topic, because the extra consumer does not improve the processing speed, and a partition can only be consumed by one consumer in a consumer group at a time.

Extensibility in code belongs to the area of design patterns, which is not discussed here.

This is the end of the introduction to "what are the concepts of Kubernetes message queuing". Thank you for reading. If you want to know more about the industry, you can follow the industry information channel. The editor will update different knowledge points for you every day.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.