我已经将下面的代码拼凑在一起,不做任何复杂的事情--只需创建一个byte[]变量,将其写入Cassandra (v1.2,通过新的Datastax CQL库)中的blob字段,然后再次读出它。
当我把它放进去时,它有3个元素长,当我重新读它的时候,它有84个元素长……!这意味着当我再次尝试反序列化时,我实际上想做的事情(序列化Java对象)失败了,并出现了org.apache.commons.lang.SerializationException: java.io.StreamCorruptedException: invalid stream header: 81000008
错误。
下面是一些示例代码,演示了我的问题:
import java.nio.ByteBuffer;
import org.apache.commons.lang.SerializationUtils;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.Metadata;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
public class TestCassandraSerialization {
private Cluster cluster;
private Session session;
public TestCassandraSerialization(String node) {
connect(node);
}
private void connect(String node) {
cluster = Cluster.builder().addContactPoint(node).build();
Metadata metadata = cluster.getMetadata();
System.out.printf("Connected to %s\n", metadata.getClusterName());
for (Host host: metadata.getAllHosts()) {
System.out.printf("Datacenter: %s; Host: %s; Rack: %s\n",
host.getDatacenter(), host.getAddress(), host.getRack());
}
session = cluster.connect();
}
public void setUp() {
session.execute("CREATE KEYSPACE test_serialization WITH replication = {'class':'SimpleStrategy', 'replication_factor':1};");
session.execute("CREATE TABLE test_serialization.test_table (id text PRIMARY KEY, data blob)");
}
public void tearDown() {
session.execute("DROP KEYSPACE test_serialization");
}
public void insertIntoTable(String key, byte[] data) {
PreparedStatement statement = session.prepare("INSERT INTO test_serialization.test_table (id,data) VALUES (?, ?)");
BoundStatement boundStatement = new BoundStatement(statement);
session.execute(boundStatement.bind(key,ByteBuffer.wrap(data)));
}
public byte[] readFromTable(String key) {
String q1 = "SELECT * FROM test_serialization.test_table WHERE id = '"+key+"';";
ResultSet results = session.execute(q1);
for (Row row : results) {
ByteBuffer data = row.getBytes("data");
return data.array();
}
return null;
}
public static boolean compareByteArrays(byte[] one, byte[] two) {
if (one.length > two.length) {
byte[] foo = one;
one = two;
two = foo;
}
// so now two is definitely the longer array
for (int i=0; i<one.length; i++) {
//System.out.printf("%d: %s\t%s\n", i, one[i], two[i]);
if (one[i] != two[i]) {
return false;
}
}
return true;
}
public static void main(String[] args) {
TestCassandraSerialization tester = new TestCassandraSerialization("localhost");
try {
tester.setUp();
byte[] dataIn = new byte[]{1,2,3};
tester.insertIntoTable("123", dataIn);
byte[] dataOut = tester.readFromTable("123");
System.out.println(dataIn);
System.out.println(dataOut);
System.out.println(dataIn.length); // prints "3"
System.out.println(dataOut.length); // prints "84"
System.out.println(compareByteArrays(dataIn, dataOut)); // prints false
String toSave = "Hello, world!";
dataIn = SerializationUtils.serialize(toSave);
tester.insertIntoTable("toSave", dataIn);
dataOut = tester.readFromTable("toSave");
System.out.println(dataIn.length); // prints "20"
System.out.println(dataOut.length); // prints "104"
// The below throws org.apache.commons.lang.SerializationException: java.io.StreamCorruptedException: invalid stream header: 81000008
String hasLoaded = (String) SerializationUtils.deserialize(dataOut);
System.out.println(hasLoaded);
} finally {
tester.tearDown();
}
}
}
看起来像是正确的东西进入了数据库:
cqlsh:flight_cache> select * from test_serialization.test_table;
id | data
--------+--------------------------------------------
123 | 0x010203
toSave | 0xaced000574000d48656c6c6f2c20776f726c6421
cqlsh:flight_cache>
因此,当读取而不是写入二进制数据时,它看起来像是一个错误。有谁能给我指点一下我哪里做错了吗?
https://stackoverflow.com/questions/17282361
复制相似问题