通过ByteBuffer和CQL 3将Java对象序列化为Cassandra 1.2
我拼凑了下面没有做任何复杂的代码 – 只需创建一个byte []变量，将其写入Cassandra的blob字段（v1.2，通过新的Datastax CQL库），然后将其读回再次出来。
当我把它放在它的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>
因此，在读取而不是写入二进制数据时看起来像是一个错误。 谁能给我任何关于我做错的指示？
问题几乎可以肯定是因为ByteBuffer.array（）返回的数组是完整的后备数组，但数据可能只包含在其中的一部分中。
返回的有效数据从ByteBuffer.arrayOffset（）开始，长度为ByteBuffer.remaining（）。 要获取仅包含有效数据的字节数组，请在readFromTable中使用此代码：
byte[] result = new byte[data.remaining()]; data.get(result);
那么你的数据就在结果中，你可以返回。
由于您已经使用了DataStax Java驱动程序，因此com.datastax.driver.core.utils还有一个实用程序类 ，您可以使用它：
byte[] result = Bytes.getArray(data)