import java.util.*;
import java.io.*;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class ManualConsumer{

    public static void main(String[] args) throws Exception{

        String topicName = "SupplierTopic";
        String groupName = "SupplierTopicGroup";

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093");
        props.put("group.id", groupName);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "SupplierDeserializer");
        props.put("enable.auto.commit", "false");

        KafkaConsumer<String, Supplier> consumer = null;

        try {
            consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Arrays.asList(topicName));

            while (true){
                ConsumerRecords<String, Supplier> records = consumer.poll(100);
                for (ConsumerRecord<String, Supplier>record : records){
                    System.out.println("Supplier id= " + String.valueOf(record.value().getID()) + 
                                       " Supplier  Name = " + record.value().getName() + 
                                       " Supplier Start Date = " + record.value().getStartDate().toString());
                }
                consumer.commitAsync();
            }
        }catch(Exception ex){
            ex.printStackTrace();
        }finally{
            consumer.commitSync();
            consumer.close();
        }
    }
}                           

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092,localhost:9093");

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("partitioner.class", "com.example.MyPartitioner");
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<>(props);

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "test-group");
props.setProperty("enable.auto.commit", "false");
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
//don't call consumer#subscribe()
//assigning partition-id=1
consumer.assign(Collections.singleton(new TopicPartition("topic", 1)));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2));
    for (ConsumerRecord<String, String> record : records) {
        // todo sth.
    }
    consumer.commitSync(); // enable.auto.commit 设置为 false 时必须
}

// 连接 kafka
@Bean
public KafkaAdmin admin() {
    Map<String, Object> configs = new HashMap<>();
    configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    return new KafkaAdmin(configs);
}

@Bean
public NewTopic divinerInternalStrategyOperation() {
    return TopicBuilder.name("example-topic")
            .partitions(1)
            .replicas(1)
            .compact()
            .build();
}

@KafkaListener(id = "consumer-group", topics = "diviner_internal_strategy_operation")
public void listen(String in) {
    // ...
}

@Autowired
private KafkaTemplate<Object, Object> template;
 
template.send(key, value);

package pub.wii.cook.springboot.cache;

import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.collect.Lists;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.cache.caffeine.CaffeineCacheManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.TimeUnit;

@EnableCaching
@Configuration
public class CustomCacheManager {
    private static final int CACHE_CAP = 100;
    public static final String CACHE_NAME = "sample";

    @Bean(name = CACHE_NAME)
    CacheManager cacheManager() {
        CaffeineCacheManager cm = new CaffeineCacheManager();
        cm.setCaffeine(Caffeine.newBuilder().expireAfterAccess(1, TimeUnit.MINUTES)
                .recordStats()
                .initialCapacity(CACHE_CAP)
                .maximumSize(CACHE_CAP));
        cm.setCacheNames(Lists.newArrayList(CACHE_NAME));
        return cm;
    }
}

package pub.wii.cook.springboot.cache;

import org.springframework.cache.interceptor.KeyGenerator;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.Arrays;

@Component
public class CustomKeyGenerator implements KeyGenerator {
    @SuppressWarnings("NullableProblems")
    @Override
    public Object generate(Object o, Method method, Object... objects) {
        return o.getClass().getSimpleName() + ":" + method.getName() + ":" + Arrays.toString(objects);
    }
}

package pub.wii.cook.springboot.cache;

import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.UUID;

import static pub.wii.cook.springboot.cache.CustomCacheManager.CACHE_NAME;

@Service
public class CacheableService {
    private static final Random random = new Random();

    @Cacheable(
            cacheManager = CACHE_NAME,
            cacheNames = CACHE_NAME,
            keyGenerator = "customKeyGenerator"
    )
    public List<Object> cache(String key) {
        List<Object> res = new ArrayList<>();
        int size = random.nextInt(10) + 1;
        for (int i = 0; i < size; ++i) {
            res.add(UUID.randomUUID().toString());
        }
        return res;
    }
}

package pub.wii.cook.springboot.controller;

import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import pub.wii.cook.springboot.cache.CacheableService;

import javax.annotation.Resource;
import java.util.List;

@RestController
@RequestMapping("cache")
public class CacheController {

    @Resource
    CacheableService cacheableService;

    @RequestMapping(value = "get",
            method = RequestMethod.GET,
            produces = MediaType.APPLICATION_JSON_VALUE)
    @ResponseBody
    public ResponseEntity<List<Object>> get(@RequestParam("key") String key) {
        return ResponseEntity.ok(cacheableService.cache(key));
    }
}

@Service
public class CacheableService {
    private static final Random random = new Random();

    @Cacheable(
            cacheManager = CACHE_NAME,
            cacheNames = CACHE_NAME,
            keyGenerator = "customKeyGenerator",
            unless = "#result == null or #result.size() == 0"
    )
    public List<Object> cache(String key) {
        List<Object> res = new ArrayList<>();
        int size = random.nextInt(10) + 1;
        for (int i = 0; i < size; ++i) {
            res.add(UUID.randomUUID().toString());
        }
        return res;
    }
}

package pub.wii.cook.springboot.cache;

import com.google.common.collect.Lists;
import lombok.SneakyThrows;
import org.junit.jupiter.api.Test;
import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.cache.interceptor.KeyGenerator;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import pub.wii.cook.springboot.config.CookSpringBootConfiguration;

import javax.annotation.Resource;

import java.util.ArrayList;
import java.util.List;

import static org.junit.jupiter.api.Assertions.*;
import static pub.wii.cook.springboot.cache.CustomCacheManager.CACHE_NAME;

@ContextConfiguration(classes = CookSpringBootConfiguration.class)
@SpringJUnitConfig(classes = CookSpringBootConfiguration.class)
class CacheableServiceTest {

    interface ICache {
        List<Integer> cacheWithEmpty(List<Integer> echo);

        List<Integer> cacheWithoutEmpty(List<Integer> echo);
    }

    @Configuration
    @EnableCaching
    static class Config {
        static class ICacheImpl implements ICache {

            @Cacheable(
                    cacheManager = CACHE_NAME,
                    cacheNames = CACHE_NAME,
                    keyGenerator = "customKeyGenerator"
            )
            @Override
            public List<Integer> cacheWithEmpty(List<Integer> echo) {
                return echo;
            }

            @Cacheable(
                    cacheManager = CACHE_NAME,
                    cacheNames = CACHE_NAME,
                    keyGenerator = "customKeyGenerator",
                    unless = "#result == null or #result.size() == 0"
            )
            @Override
            public List<Integer> cacheWithoutEmpty(List<Integer> echo) {
                return echo;
            }
        }

        @Bean
        ICache iCache() {
            return new ICacheImpl();
        }
    }

    @Resource(name = CACHE_NAME)
    CacheManager cacheManager;

    @Resource
    ICache iCache;

    @SuppressWarnings("ConstantConditions")
    @SneakyThrows
    @Test
    void cache() {
        List<Integer> nonEmpty = Lists.newArrayList(1, 2, 3);
        List<Integer> empty = new ArrayList<>();
        List<Integer> nil = null;
        Cache cache = cacheManager.getCache(CACHE_NAME);
        KeyGenerator kg = new CustomKeyGenerator();
        assertNotNull(cache);

        iCache.cacheWithEmpty(nonEmpty);
        iCache.cacheWithEmpty(empty);
        iCache.cacheWithEmpty(nil);
        iCache.cacheWithoutEmpty(nonEmpty);
        iCache.cacheWithoutEmpty(empty);
        iCache.cacheWithoutEmpty(nil);

        assertEquals(cache.get(genKeyWithEmpty(kg, nonEmpty)).get(), nonEmpty);
        assertEquals(cache.get(genKeyWithEmpty(kg, empty)).get(), empty);
        assertEquals(cache.get(genKeyWithEmpty(kg, nil)).get(), nil);
        assertEquals(cache.get(genKeyWithoutEmpty(kg, nonEmpty)).get(), nonEmpty);
        assertEquals(cache.get(genKeyWithoutEmpty(kg, empty)), nil);
        assertEquals(cache.get(genKeyWithoutEmpty(kg, nil)), nil);
    }

    @SneakyThrows
    Object genKeyWithEmpty(KeyGenerator kg, List<Integer> arg) {
        // 第一个参数不要用 iCache, iCache 是通过反射机制设置的对象, 有可能是一个 Proxy
        // 获取 class name 的时候可能会得到奇怪的值, 导致 key 匹配不上
        return kg.generate(new Config.ICacheImpl(),
                ICache.class.getMethod("cacheWithEmpty", List.class), arg);
    }

    @SneakyThrows
    Object genKeyWithoutEmpty(KeyGenerator kg, List<Integer> arg) {
        // 第一个参数不要用 iCache, iCache 是通过反射机制设置的对象, 有可能是一个 Proxy
        // 获取 class name 的时候可能会得到奇怪的值, 导致 key 匹配不上
        return kg.generate(new Config.ICacheImpl(),
                ICache.class.getMethod("cacheWithoutEmpty", List.class), arg);
    }
}

package pub.wii.cook.springboot.config;

import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.context.annotation.aspectj.EnableSpringConfigured;

@EnableSpringConfigured
@EnableAspectJAutoProxy
@ComponentScan(basePackages = {"pub.wii.cook"}, lazyInit = true)
public class CookSpringBootConfiguration {
}