Produzindo mensagens com Kafkae 모드 레지스트리

카프카 아트센터는 아파치 카프카의 예술 형식에 공공 서비스를 제공하는 데 주력하는 기구다.
타레법은 간단한 파시리 섬으로, 다다도의 대체량을 정확하게 묘사할 수 있는 매우 중요한 체계이다.
Pensando nisso o Linked In desenvolveu um a ferramenta para comunica ço de mensagens ass çncronas, levando em considera ç o contextos on de grande volume de dados seja algo impactante, posteriormente se torndo opeache pela Apache Kafkaéuma ferramenta robusta, rápida escal álle.
현재 많은 사람들이 그들의 수요를 충족시키기 위해 새로운 방법을 찾고 있으며, 카프카의 특수한 디자인과 형식적인 실용성, 그리고 카프카 대성당의 디자인과 생산에 대한 수요를 찾고 있다.
카프카 제품은 아티고 criaremos 회사, Também iremos mostrar 회사의 제품은 효과적인 제품으로 실용적인 봄 부츠입니다.
Também n vamos nos alongar muito nos Conception sobre o que éo Kafka e suas Specificiateds caso tenha Interest sugerimos os seguintes artigos):

Introdução ao Kafka pela Apache

O que é Kafka pela Confluent

O que é Kafka pela RedHat

Criando o ambiente

카프카의 인프라 시설 건설에서 카프카와 오트로스 서비스 센터의 본보기로서 카프카와 오트로스 서비스 센터는 당신의 수요를 만족시키기 위해 서비스를 제공할 것입니다.우리 부두 일꾼과 부두 일꾼들은 레만타와 노스소 ambiente 호텔을 구성했다.
카프카의 기능성 디자인에서 환경 보호에 사용되는 이미지와 환경 보호에 사용되는 이미지, 부두에 사용되는 이미지와 환경 설정에 사용되는 제어 시스템이 있다.
오, 노소 도크.yml ficaráassim:

version: '2'

services:
  # this is our kafka cluster.
  kafka-cluster:
    image: landoop/fast-data-dev:cp3.3.0
    environment:
      ADV_HOST: 127.0.0.1         # Change to 192.168.99.100 if using Docker Toolbox
      RUNTESTS: 0                 # Disable Running tests so the cluster starts faster
      FORWARDLOGS: 0              # Disable running 5 file source connectors that bring application logs into Kafka topics
      SAMPLEDATA: 0               # Do not create sea_vessel_position_reports, nyc_yellow_taxi_trip_data, reddit_posts topics with sample Avro records.
    ports:
      - 2181:2181                 # Zookeeper
      - 3030:3030                 # Landoop UI
      - 8081-8083:8081-8083       # REST Proxy, Schema Registry, Kafka Connect ports
      - 9581-9585:9581-9585       # JMX Ports
      - 9092:9092                 # Kafka Broker

부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두 부두yml e executar:

docker-compose up -d

Após fazer isso podemos acessarhttp://localhost:3030/e conseguiremos abrir o dashboard que a Landoop Disponiliza e teremos Algom parecido com isso:

모드 레지스트리

프로젝트에 대해 평가를 하고 모델 등록의 중요성을 확보한다.
Kafka envia e recebe mensagens porém nãO faz validaãO sobre O que estásendo Envivia O recebido a a a aplicaãO Consumidor a tent Realization a desserializaãO daãO Mensage e caso contracto da aplicaãO Consumidor a a a a a a a aéa a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a.에비타는 융합된 모델 등록 센터로 효과적인 계약과 거래 협의이다.
grosso modo Schema Registry는 estásendo enviada por uma aplica çoécompat 啊vel의 전략을 검증했습니다.Podemos usar vários formatos de arquivos para criar os nossos schema como XML、CSV、JSON mas aqui usaremos Apache Avro que um formato desenvolvido para criaço de schema com tipagem.
O 카프카 아파트 모드 등록표 구성 요소:

아프로

파라코메로는 알티고 소브레 프로젝트 책임자Spring Security com JWT다.
중앙적립금위원회(CPF) 구성원으로 구성된 위원회다.
Vamos come çar criando o o nosso Avro, dentroda 파스타 자원/Avro criamos o arquivo 납세자-v1.avsc contendo nosso 모드:

{
     "type": "record",
     "namespace": "com.irs.register.avro.taxpayer",
     "name": "TaxPayer",
     "version": "1",
     "fields": [
       { "name": "name", "type": "string", "doc": "Name of TaxPayer" },
       { "name": "document", "type": "string", "doc": "Document of TaxPayer" },
       { "name": "situation", "type": "boolean", "default": false, "doc": "Legal situation of TaxPayer" }
     ]
}

O nosso Avro contém os metadados de type,namespace,name e version.Também adicionamos os campos da nossa entidade에는 배열 필드가 없습니다. conseguimos além에는 colocar outros atributos como tipagem com 유형 e valores padr padr [compo campo default]라는 이름이 없기 때문입니다.
와모스 아디시오나르는 독립된 프로젝트로서

<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>1.10.1</version>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
</dependency>

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>5.3.0</version>
</dependency>

그는 우수한 자바 통신원이다.

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
        </plugin>

        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
        </plugin>
        <!--for specific record -->
        <plugin>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-maven-plugin</artifactId>
            <version>1.10.1</version>
            <executions>
                <execution>
                    <phase>generate-sources</phase>
                    <goals>
                        <goal>schema</goal>
                        <goal>protocol</goal>
                        <goal>idl-protocol</goal>
                    </goals>
                    <configuration>
                        <sourceDirectory>${project.basedir}/src/main/resources/avro</sourceDirectory>
                        <stringType>String</stringType>
                        <createSetters>false</createSetters>
                        <enableDecimalLogicalType>true</enableDecimalLogicalType>
                        <fieldVisibility>private</fieldVisibility>
                    </configuration>
                </execution>
            </executions>
        </plugin>
        <!--force discovery of generated classes -->
        <plugin>
            <groupId>org.codehaus.mojo</groupId>
            <artifactId>build-helper-maven-plugin</artifactId>
            <version>3.0.0</version>
            <executions>
                <execution>
                    <id>add-source</id>
                    <phase>generate-sources</phase>
                    <goals>
                        <goal>add-source</goal>
                    </goals>
                    <configuration>
                        <sources>
                            <source>target/generated-sources/avro</source>
                        </sources>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

Com isso feito podemos rodar o comando mvn 생성원 e nossa classe serágerada em 목표/생성원/avro/납세자.자바.

카프카

정확한 배치는 카프카의 핵심이다.O카프카는 계속해서 고객을 위해 맞춤형 배치를 하고 부동산을 통해 고객에게 유연한 서비스를 제공한다.
Vamos criar a nossa classe de configura ço dos nossos 호텔:

@Configuration
@ConfigurationProperties(
    prefix = "kafka"
)
@Data
@NoArgsConstructor
public class KafkaProperties {

    private List<String> bootstrapServers;
    private String acksConfig;
    private String retriesConfig;
    private Class<?> keySerializer = StringSerializer.class;
    private Class<?> valueSerializer = KafkaAvroSerializer.class;
    private String schemaRegistryUrl;

}

E에는 nosso arquivo 응용 프로그램이 없습니다.yml:

kafka:
  bootstrapServers: 127.0.0.1:9092 
  acksConfig: all
  retriesConfig: 10
  schemaRegistryUrl: "http://127.0.0.1:8081"

카프카드도 없고, 패턴 등록도 없고, 연속적인 디자인 방안도 없고, 환경 확인 방안도 없고,환경 보호의 정보원으로서 우리는 파하람 지역의 환경 보호에 대해 재평가를 해야 한다.수출 상품의 구조는 일종의 신제품이다.
카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카 카프카

@Configuration
public class MessagingConfigTaxPayer implements MessagingConfigPort<TaxPayer> {

    @Autowired
    private KafkaProperties kafkaProperties;

    @Bean(name = "taxpayerProducer")
    @Override
    public KafkaProducer<String, TaxPayer> configureProducer() {

        Properties properties = new Properties();

        properties.put(BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
        properties.put(ACKS_CONFIG, kafkaProperties.getAcksConfig());
        properties.put(RETRIES_CONFIG, kafkaProperties.getRetriesConfig());
        properties.put(KEY_SERIALIZER_CLASS_CONFIG, kafkaProperties.getKeySerializer());
        properties.put(VALUE_SERIALIZER_CLASS_CONFIG, kafkaProperties.getValueSerializer());
        properties.put(SCHEMA_REGISTRY_URL_CONFIG, kafkaProperties.getSchemaRegistryUrl());

        return new KafkaProducer<String, TaxPayer>(properties);

    }

}

제품 구성

Agora vamos criar o nosso produtor que implementa a interface MessagingPort que possui três métodos:

문자열 테마()

생산기록 생성 생산기록(T T)

무효 발송 (보통 dto로)

우리의 목표는 소비자의 이익을 확보하여 카프카를 차별받지 않는 납세자로 만드는 것이다.

@Service
@Slf4j
public class TaxpayerService implements MessagingPort<TaxPayer> {

    @Autowired
    @Qualifier("taxpayerProducer")
    private KafkaProducer<String, TaxPayer> producer;

    @Override
    public String topic() {
        return "taxpayer-avro";
    }

    @Override
    public ProducerRecord<String, TaxPayer> createProducerRecord(TaxPayer taxPayer) {

        return new ProducerRecord<String, TaxPayer>(this.topic(), taxPayer);

    }

    @Override
    public void send(CommonDTO taxpayerDTO) {


        TaxPayer taxPayer = TaxPayer.newBuilder().setName(((TaxpayerDTO) taxpayerDTO).getName())
                .setDocument(((TaxpayerDTO) taxpayerDTO).getDocument()).setSituation(false).build();


        producer.send(this.createProducerRecord(taxPayer), (rm, ex) -> {
            if (ex == null) {
                log.info("Data sent with success!!!");
            } else {
                log.error("Fail to send message", ex);
            }
        });

        producer.flush();
        producer.close();

    }

}

Detalhando cada método.
메토도 주제가 없고, 카프카가 없다.
Ométodo create Producer Record Receibe como par – metro O nosso 납세자는 생산 기록을 이체합니다.
Ométodo send recebe um CommonDTO, que nada mais édo que uma interface de marca çO para os DTOs da aplica çO, nele podemos ver que usamos O Builder que a 납세자 fornece, passando os dados que iremos receber no POST da API.
카프카의 디자인에서 카프카 프로듀서의 디자인과 제작 과정은 모두 간단한 검증 과정이다.아포오스 isso'아투아니즈 아모스'는 페카모스를 넘나드는 지역이다.

컨트롤러

정확히 말하면 이것은 aplicaço의 입구이고 서비스의 입구이다.

@RestController
@RequestMapping("/taxpayer")
public class TaxpayerController {

    @Autowired
    private TaxpayerService taxpayerService;

    @PostMapping
    public ResponseEntity<TaxpayerDTO> postTaxpayer(@RequestBody TaxpayerDTO taxpayer){

        taxpayerService.send(taxpayer);

        return ResponseEntity.ok(taxpayer);
    }

}

Criamos o endpoint/Taxboyer(종점/납세자)는 환경 및 서비스에 대한 응답을 제공하는 서비스를 납세자에게 제공합니다.
납세자:

@Data
public class TaxpayerDTO implements CommonDTO{

    private String name;

    private String document;

    @Override
    public String getType() {
        return "TaxPayerDTO";
    }

}

집행관

Vamos Executator a aplicaão e envirar um POST para conference o function,para envirar os dados foi utilizado oGerador de Pessoase tambéménecessário envirar o token JWT para autorizaãa esso,para saber mais sobre isso isso consultar o artigo sobreSpring Security com JWT.

우리는 하나의 계기판과 하나의 계기판으로 우리의 설계 방안을 전시할 수 있다.

cima vemos o schema criado.

우리는 환경에 관한 정보를 필요로 한다.

단말기를 통해 소비하다

Podemos produzir, consumir, criar e fazer todas, opera fes do Kafka는 터미널을, agora para fins de Examplo vamos consumir와 Mensagage que enviamos는 터미널을 통과한다.
Docker 등록 모드를 통해 Podemos fazer isso acessando nosso:

docker run -it --rm --net=host confluentinc/cp-schema-registry:3.3.1 bash

E. 소비자 akafka avro 콘솔을 사용하는 소비자:

kafka-avro-console-consumer --topic taxpayer-avro \
     --bootstrap-server localhost:9092 \
     --from-beginning \
     --property schema.registry.url=http://127.0.0.1:8081

결과:

{"name":"Luís Marcelo da Conceição","document":"216.172.648-06","situation":false}

코디고 봉트

código completo noGitHub