我正在尝试向Kafka发送一条消息,并使用spring cloud stream使用来自Kafka的相同消息。我正在使用postman将json字符串{“acctno”:“32432”,“tn”:“3234”}发送到生产者的rest控制器。我得到了json parserexception:
com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'ÿ': was expecting ('true', 'false' or 'null')
at [Source: [B@776e03af; line: 1, column: 4]
at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1581) ~[jackson-core-2.6.5.jar:2.6.5]
at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:533) ~[jackson-core-2.6.5.jar:2.6.5]
at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3451) ~[jackson-core-2.6.5.jar:2.6.5]
at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._handleUnexpectedValue(UTF8StreamJsonParser.java:2610) ~[jackson-core-2.6.5.jar:2.6.5]
at com.fasterxml.jackson.core.json.UTF8StreamJsonParser
我的代码如下:
生产商应用yaml
spring:
cloud:
stream:
bindings:
activationMsgQueue:
destination: test
contentType: application/json
生产者休息控制器
@RestController
@RequestMapping("/ActivationQueueService")
public class ActivationQueueController {
private static final Logger LOGGER = LoggerFactory
.getLogger(ActivationQueueController.class);
@Autowired
SpringCloudStreamClient producer;
@InitBinder
protected void initBinder(WebDataBinder binder) {
binder.setValidator(new ActivationDataInfoValidator());
}
@RequestMapping(method = RequestMethod.POST, value = "/sendMessage", headers = "Accept=application/json", produces = "application/json")
public void sendMessage(@RequestBody @Valid ActivationDataInfo message)
throws JsonProcessingException {
LOGGER.debug("Activation Data Request Recieved : " + message.toString());
if (message != null) {
ObjectMapper mapper = new ObjectMapper();
producer.sendMessagetoKafka(message);
LOGGER.info("Activation Data Request sent to Kafka : " + message);
}
}
}
生产商代码
@Service
@EnableBinding(MessageChannels.class)
public class SpringCloudStreamClient {
private static final Logger LOGGER = LoggerFactory
.getLogger(SpringCloudStreamClient.class);
@Autowired MessageChannels msgChannel;
public Object sendMessagetoKafka(ActivationDataInfo msg){
LOGGER.info("Sending Message : " + msg);
msgChannel.save().send(MessageBuilder.withPayload(msg).build());
return new String("Success");
}
}
用户应用程序yaml
spring:
cloud:
stream:
bindings:
input:
content-type: application/x-java-object;type=com.comcast.activation.message.vo.ActivationDataInfo
destination: test
group: prac
consumer:
headerMode: raw
enableDlq: true
resetOffsets: true
startOffset: latest
消费者代码
@EnableBinding(Sink.class)
public class LogSink {
private static Logger logger = LoggerFactory.getLogger(LogSink.class);
@ServiceActivator(inputChannel = Sink.INPUT)
public void loggerSink( ActivationDataInfo payload) throws Exception {
logger.info("Received: " + payload.getAcctNo());
}
}
域类
public class ActivationDataInfo {
private String acctNo;
private String tn;
public String getAcctNo() {
return acctNo;
}
public void setAcctNo(String acctNo) {
this.acctNo = acctNo;
}
public String getTn() {
return tn;
}
public void setTn(String tn) {
this.tn = tn;
}
是什么原因造成的?请帮忙。
1条答案
按热度按时间ohtdti5x1#
尝试删除
headerMode: raw
因为制作人也不使用它。