这篇文章上次修改于 2229 天前，可能其部分内容已经发生变化，如有疑问可询问作者。 ### kafka 发消息报错 ```java org.apache.kafka.common.errors.SerializationException: Can't convert value of class java.lang.String to class org.apache.kafka.common.serialization.ByteArraySerializer specified in value.serializer Caused by: java.lang.ClassCastException: java.lang.String cannot be cast to [B at org.apache.kafka.common.serialization.ByteArraySerializer.serialize(ByteArraySerializer.java:21) at org.apache.kafka.common.serialization.ExtendedSerializer$Wrapper.serialize(ExtendedSerializer.java:65) at org.apache.kafka.common.serialization.ExtendedSerializer$Wrapper.serialize(ExtendedSerializer.java:55) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:791) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:768) at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.send(DefaultKafkaProducerFactory.java:363) at brave.kafka.clients.TracingProducer.send(TracingProducer.java:79) at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:355) at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:182) at tv.rr.thor.bus.kafka.producer.KafkaMsgProducer.send(KafkaMsgProducer.java:34) at tv.rr.thor.bus.kafka.producer.KafkaMsgProducer$$FastClassBySpringCGLIB$$65ccb95e.invoke(<generated>) at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204) at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746) at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) at org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint.proceed(MethodInvocationProceedingJoinPoint.java:88) at org.springframework.cloud.sleuth.instrument.async.TraceAsyncAspect.traceBackgroundThread(TraceAsyncAspect.java:65) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethodWithGivenArgs(AbstractAspectJAdvice.java:644) at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:633) at org.springframework.aop.aspectj.AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:70) at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185) at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:92) at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185) at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688) at tv.rr.thor.bus.kafka.producer.KafkaMsgProducer$$EnhancerBySpringCGLIB$$d0f2998e.send(<generated>) at tv.rr.thor.bus.ThorBusApplicationTests.testKafka(ThorBusApplicationTests.java:29) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50) at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12) at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47) at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17) at org.springframework.test.context.junit4.statements.RunBeforeTestExecutionCallbacks.evaluate(RunBeforeTestExecutionCallbacks.java:73) at org.springframework.test.context.junit4.statements.RunAfterTestExecutionCallbacks.evaluate(RunAfterTestExecutionCallbacks.java:83) at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:75) at org.springframework.test.context.junit4.statements.RunAfterTestMethodCallbacks.evaluate(RunAfterTestMethodCallbacks.java:86) at org.springframework.test.context.junit4.statements.SpringRepeat.evaluate(SpringRepeat.java:84) at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325) at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:251) at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97) at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290) at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71) at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288) at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58) at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268) at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61) at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70) at org.junit.runners.ParentRunner.run(ParentRunner.java:363) at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190) ``` ### 代码如下 ```java @Component public class KafkaMsgProducer { @Autowired private KafkaTemplate<Long, String> kafkaTemplate; private static final Logger log = LoggerFactory.getLogger(KafkaMsgProducer.class); @Async public void send(String topic, Object msg) { ListenableFuture<SendResult<Long, String>> future = kafkaTemplate.send(topic, JSON.toJSONString(msg)); future.addCallback(new ListenableFutureCallback<SendResult<Long, String>>() { @Override public void onFailure(Throwable throwable) { log.warn("Sending message to Kafka failed, topic {}, msg: {}", topic, JSON.toJSONString(msg)); } @Override public void onSuccess(SendResult<Long, String> season) { log.info("Sending message to Kafka finished, topic {}, msg: {}", topic, JSON.toJSONString(msg)); } }); } } ``` ### 原因分析：java value的序列化和反序列化没有配置，配置如下 ```yaml spring: kafka: producer: value-serializer: org.apache.kafka.common.serialization.StringSerializer # key-serializer: org.apache.kafka.common.serialization.LongSerializer consumer: value-deserializer: org.apache.kafka.common.serialization.StringDeserializer ``` ### 还有有的项目不需要配置，怀疑是版本问题，默认序列化器有的是String,有的不是
没有评论