Dolphin的博客

发表于 | 分类于 |

kafka允许通过seek(TopicPartition，long)指定新的位置，或者seekToBeginning，seekToEnd定位到最早或最近的offset。注意seek重置offsets只对当前消费者起作用，它并不会触发consumer的rebalance，或者影响其他消费者的fetchOffsets。在大多数情况下，消费者消费记录只是简单地从一开始到结束，并且定时地提交它的位置(不管是自动的还是手动的)。不过新的API也允许消费者手动控制它的位置，消费者可以在一个partition钟随意地往前或者往后移动位置。这就意味着消费者可以重新消费旧的记录(多次读取相同的记录)，或者直接跳到最近的记录，忽略掉中间的记录。

阅读全文 »
Dolphin的博客

发表于 | 分类于 |

简介

Redis是一个开源的使用ANSI C语言编写、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库，并提供多种语言的API，其实当前最热门的NoSQL数据库之一，NoSQL还包括了Memcached和mongodb。

阅读全文 »
Dolphin的博客

发表于 | 分类于 |

启动kafka：

1
./kafka-server-start.sh ../config/server.properties &

创建主题：

1
./kafka-topics.sh --zookeeper localhost:2181 --create --topic test1 --partitions 1 --replication-factor 1 --config max.message.bytes=64000 --config flush.messages=1

查看所有主题：

1
./kafka-topics.sh --list --zookeeper 192.168.24.11:2181
阅读全文 »
Dolphin的博客

发表于 | 分类于 |

这里的Kafka的版本是0.9.0.1，查看所有Topics和分区：

1
./kafka-topics.sh --describe --zookeeper 192.168.244.11:2181

只查看所有Topic：

1
./kafka-topics.sh --list --zookeeper 192.168.24.244:2181

删除主题：

1
./kafka-topics.sh --zookeeper 192.168.24.244:2181 --delete --topic 0085000
阅读全文 »
Dolphin的博客

发表于 | 分类于 |

Invalid partition given with record

在Kafka生产者里写入消息时，提示写入失败，详细的错误信息如下所示：

1
2
3
4
5
6
7
8
9
[ERROR]-[2016年-10月-28日16:17:35.083]-[Thread-25]-[com.zw.socket.service.kafka.producer.ClientKafkaProducer]-{生产者发送消息出错}java.lang.IllegalArgumentException: Invalid partition given with record: 1 is not in the range [0...1].
	at org.apache.kafka.clients.producer.KafkaProducer.partition(KafkaProducer.java:671) ~[kafka-clients-0.9.0.1.jar:?]
	at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:430) ~[kafka-clients-0.9.0.1.jar:?]
	at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:339) ~[kafka-clients-0.9.0.1.jar:?]
	at com.zw.socket.service.kafka.producer.ClientKafkaProducer.sendMessage(ClientKafkaProducer.java:102) [classes/:?]
	at com.zw.socket.service.handler.common.CommonCommandHandler.commonMessageWriteIntoKafka(CommonCommandHandler.java:95) [classes/:?]
	at com.zw.socket.service.handler.device.DeviceMessageHandler.sendRegisterResult(DeviceMessageHandler.java:175) [classes/:?]
	at com.zw.socket.service.handler.device.DeviceMessageHandler.isAllowRegisted(DeviceMessageHandler.java:161) [classes/:?]
	at com.zw.socket.service.handler.device.DeviceMessageHandler.saveDeviceInfo(DeviceMessageHandler.java:216) [classes/:?]
阅读全文 »