如何底层调用最快地复制OPC数据到关系数据库

int main(int argc, CHAR* argv[]) {
	
	CoInitializeEx(NULL, COINIT_MULTITHREADED);

	{
		CDataSource dataSource;
		CSession session;

		const WCHAR szUDLFile[] = L"OPCDA.udl";

		HRESULT hr = dataSource.OpenFromFileName(szUDLFile);

		if (FAILED(hr)) {

			printf("OpenFromFileName() failed\n");
			goto END;
		}

		hr = session.Open(dataSource);

		if (FAILED(hr))
		{
			printf("Open() failed\n");
			dataSource.Close();
			goto END;
		}

		CLSID cidOpcServer;

		if (FAILED(listServers(cidOpcServer)))
		{
			printf("listServers() failed\n");
			dataSource.Close();
			goto END;
		}

		if (FAILED(DA(cidOpcServer, session))) {
			printf("DA() failed\n");
			dataSource.Close();
			goto END;
		}

		printf("\nretrieving rows from database...\n\n");

		displayResult(session);

		dataSource.Close();
	}

	system("pause");

END:
	CoUninitialize();
	return(EXIT_SUCCESS);
}

[oledb]
; Everything after this line is an OLE DB initstring
Provider=MSOLEDBSQL19.1;Integrated Security=SSPI;Persist Security Info=False;Initial Catalog=TEST;Data Source=localhost;Use Encryption for Data=Optional;

HRESULT OpenFromFileName(_In_z_ LPCOLESTR szFileName) throw()
{
	CComPtr<IDataInitialize> spDataInit;
	CComHeapPtr<OLECHAR>     spszInitString;

	HRESULT hr = CoCreateInstance(__uuidof(MSDAINITIALIZE), NULL, CLSCTX_INPROC_SERVER,
		__uuidof(IDataInitialize), (void**)&spDataInit);
	if (FAILED(hr))
		return hr;

	hr = spDataInit->LoadStringFromStorage(szFileName, &spszInitString);
	if (FAILED(hr))
		return hr;

	return OpenFromInitializationString(spszInitString);
}
// Open the datasource specified by the passed initialization string
HRESULT OpenFromInitializationString(
	_In_z_ LPCOLESTR szInitializationString,
	_In_ bool fPromptForInfo = false) throw()
{
	CComPtr<IDataInitialize> spDataInit;

	HRESULT hr = CoCreateInstance(__uuidof(MSDAINITIALIZE), NULL, CLSCTX_INPROC_SERVER,
		__uuidof(IDataInitialize), (void**)&spDataInit);
	if (FAILED(hr))
		return hr;

	hr = spDataInit->GetDataSource(NULL, CLSCTX_INPROC_SERVER, szInitializationString,
		__uuidof(IDBInitialize), (IUnknown**)&m_spInit);
	if (FAILED(hr))
		return hr;

	if( fPromptForInfo )
	{
		CComPtr<IDBProperties> spIDBProperties;
		hr = m_spInit->QueryInterface( &spIDBProperties );

		DBPROP rgProperties[1];
		DBPROPSET rgPropertySets[1];

		VariantInit(&rgProperties[0].vValue);
		rgProperties[0].dwOptions = DBPROPOPTIONS_REQUIRED;
		rgProperties[0].colid = DB_NULLID;
		rgProperties[0].dwPropertyID = DBPROP_INIT_PROMPT;
		rgProperties[0].vValue.vt = VT_I2;
		rgProperties[0].vValue.lVal = DBPROMPT_COMPLETEREQUIRED;

		rgPropertySets[0].rgProperties = rgProperties;
		rgPropertySets[0].cProperties = 1;
		rgPropertySets[0].guidPropertySet = DBPROPSET_DBINIT;

		hr = spIDBProperties->SetProperties( 1, rgPropertySets );
		if (FAILED(hr))
			return hr;
	}

	return m_spInit->Initialize();
}

HRESULT Open(
	_In_ const CDataSource& ds,
	_Inout_updates_opt_(ulPropSets) DBPROPSET *pPropSet = NULL,
	_In_ ULONG ulPropSets = 0) throw()
{
	CComPtr<IDBCreateSession> spSession;

	// Check we have connected to the database
	ATLASSERT(ds.m_spInit != NULL);

	HRESULT hr = ds.m_spInit->QueryInterface(__uuidof(IDBCreateSession), (void**)&spSession);
	if (FAILED(hr))
		return hr;

	hr = spSession->CreateSession(NULL, __uuidof(IOpenRowset), (IUnknown**)&m_spOpenRowset);

	if( pPropSet != NULL && SUCCEEDED(hr) && m_spOpenRowset != NULL )
	{
		// If the user didn't specify the default parameter, use one
		if (pPropSet != NULL && ulPropSets == 0)
			ulPropSets = 1;

		CComPtr<ISessionProperties> spSessionProperties;
		hr = m_spOpenRowset->QueryInterface(__uuidof(ISessionProperties), (void**)&spSessionProperties);
		if(FAILED(hr))
			return hr;

		hr = spSessionProperties->SetProperties( ulPropSets, pPropSet );
	}
	return hr;
}

HRESULT listServers(CLSID& cidOpcServer)
{
	ULONG fetched = 0;
	HRESULT hr = S_OK;
	CComHeapPtr<OLECHAR> bsProgID, lpszUserType, lpszVerIndProgID;

	CATID arrcatid[3] = { NULL };
	arrcatid[0] = __uuidof(CATID_OPCDAServer10);
	arrcatid[1] = __uuidof(CATID_OPCDAServer20);
	arrcatid[2] = __uuidof(CATID_OPCDAServer30);

	CComPtr<IOPCServerList2> spIOPCServerList2;

	if (FAILED(hr = spIOPCServerList2.CoCreateInstance(__uuidof(OpcServerList), spIOPCServerList2, CLSCTX_ALL)))
	{
		printf("CoCreateInstance() for IOPCServerList2 failed\n");
		return hr;
	}

	CComPtr<IOPCEnumGUID> spEnum;

	hr = spIOPCServerList2->EnumClassesOfCategories(sizeof arrcatid / sizeof CATID, arrcatid, 0, NULL, &spEnum);

	if (spEnum.p)
	{
		while ((hr = spEnum->Next(1, &cidOpcServer, &fetched)) == S_OK)
		{
			hr = spIOPCServerList2->GetClassDetails(cidOpcServer, &bsProgID, &lpszUserType, &lpszVerIndProgID);

			if (FAILED(hr)) {
				_tprintf(_T("GetClassDetails() failed\n"));
				return hr;
			}

			break;
		}
	}

	return hr;
}

HRESULT DA(CLSID& cidOpcServer, CSession& session) {

	CComPtr<IOPCServer> pIOPCServer;

	HRESULT hr = pIOPCServer.CoCreateInstance(cidOpcServer, pIOPCServer, CLSCTX_ALL);

	if (FAILED(hr)) {
		printf("CoCreateInstance() for IOPCServer failed\n");
		return E_FAIL;
	}

	DWORD dwRevisedUpdateRate = 0;
	OPCHANDLE hGroup = 0;
	CComPtr<IOPCItemMgt> pOPCItemMgt;

	 hr = pIOPCServer->AddGroup(L"", TRUE, 1000, NULL, NULL, NULL, LOCALE_SYSTEM_DEFAULT, &hGroup, &dwRevisedUpdateRate, __uuidof(IOPCItemMgt), (LPUNKNOWN*)&pOPCItemMgt);

	 if (FAILED(hr)) {
		 printf("AddGroup() failed\n");
		 return E_FAIL;
	 }
	 
	 DataCallback* pDataCallback = new DataCallback(session);
	 pDataCallback->AddRef();
	 DWORD m_dwCookie;
	 AtlAdvise(pOPCItemMgt, pDataCallback, __uuidof(IOPCDataCallback), &m_dwCookie);

	 hr = addItems(pOPCItemMgt);

	 if (FAILED(hr)) {
		 printf("addItems() failed\n");
		 return E_FAIL;
	 }

	 printf("\npress any key to complete inserting rows to database\n");
	 getchar();

	 AtlUnadvise(pOPCItemMgt, __uuidof(IOPCDataCallback), m_dwCookie);
	 pDataCallback->Release();

	return S_OK;
}

STDMETHODIMP OnDataChange(
	DWORD       dwTransid,
	OPCHANDLE   hGroup,
	HRESULT     hrMasterquality,
	HRESULT     hrMastererror,
	DWORD       dwCount,
	OPCHANDLE* phClientItems,
	VARIANT* pvValues,
	WORD* pwQualities,
	FILETIME* pftTimeStamps,
	HRESULT* pErrors
)
{
	CCommand<CManualAccessor> command;

	CComVariant vVariant[4];

	vVariant[0].vt = VT_BSTR;
	vVariant[1].vt = VT_R4;
	vVariant[2].vt = VT_DATE;
	vVariant[3].vt = VT_UINT;

	hr = command.CreateParameterAccessor(4, vVariant, sizeof vVariant); 

	if (FAILED(hr)) {
		printf("command.CreateParameterAccessor() failed");
		return hr;
	}

	for (DWORD ii = 0; ii < dwCount; ii++)
	{
		CComVariant vValue;
		WORD quality = pwQualities[ii] & OPC_QUALITY_MASK;

		COleDateTime oleTime = COleDateTime(pftTimeStamps[ii]);
		SYSTEMTIME st;
		oleTime.GetAsSystemTime(st);

		if (phClientItems[ii] == 0)
			CComBSTR("Random.Int1").CopyTo(&vVariant[0].bstrVal);
		if (phClientItems[ii] == 1)
			CComBSTR("Random.Int2").CopyTo(&vVariant[0].bstrVal);
		else if (phClientItems[ii] == 2)
			CComBSTR("Random.Real8").CopyTo(&vVariant[0].bstrVal);

		vVariant[1].fltVal = (FLOAT)pvValues[ii].dblVal;
		vVariant[2].date = oleTime;
		vVariant[3].iVal = quality;

		command.m_nCurrentParameter = 0;

		command.AddParameterEntry(1, DBTYPE_BSTR, NULL, &vVariant[0].bstrVal);
		command.AddParameterEntry(2, DBTYPE_R4, NULL, &vVariant[1].fltVal);
		command.AddParameterEntry(3, DBTYPE_DATE, NULL, &vVariant[2].date);
		command.AddParameterEntry(4, DBTYPE_UI2, NULL, &vVariant[3].iVal);
		
		/*
		This is not the most efficient and fastest way to insert a row to database due to query building/parsing and commit each time.
		To bulk insert, interface of IRowsetFastLoad has to be used and it is quite different from this code example.
		Contact developer to have a code example using IRowsetFastLoad, so you can completely understand the big difference between 
		IDBInitialize and IDataInitialize interfaces when trying to get a pointer to IRowsetFastLoad.
		*/

		hr = command.Open(session, "insert into OPCDA (Tag, Value, Time, Quality) Values (?,?,?,?)", NULL, NULL);

		if (FAILED(hr)) {
			printf("command.Open() failed");
			break;
		}
		else
			printf("\nOnDataChange: %S (%f, %s.%d, %s)", vVariant[0].bstrVal, vVariant[1].fltVal, oleTime.Format("%F %T").GetString(), st.wMilliseconds, quality == OPC_QUALITY_GOOD ? "good" : "bad");

		SysFreeString(vVariant[0].bstrVal);
	}
	
	return hr;
}

void displayResult(CSession &session) {

	CCommand<CManualAccessor> command;

	const USHORT uColumns = 4;
	CComVariant vValues[uColumns]{};

	HRESULT hr = command.CreateAccessor(uColumns, vValues, sizeof vValues);

	if (FAILED(hr))
	{
		printf("CreateAccessor() failed\n");
		return;
	}

	for (ULONG l = 0; l < uColumns; l++)
	{
		command.AddBindEntry(l + 1, DBTYPE_VARIANT, NULL, &vValues[l], NULL, NULL);
	}

	hr = command.Open(session, "select * from OPCDA", NULL, NULL);

	if (FAILED(hr))
	{
		printf("command.Open() failed\n");
		return;
	}

	ULONG count = 0;

	while (command.MoveNext() == S_OK) {

		CComVariant* pBind = (CComVariant*)command.m_pBuffer;
		
		count++;

		COleDateTime dateTime(pBind[2].date);
		printf("%S (%f, %s, %s)\n", pBind[0].bstrVal, pBind[1].fltVal, dateTime.Format("%F %T").GetString(), pBind[3].iVal == OPC_QUALITY_GOOD ? "good" : "bad");;
	}

	printf("\nTotal rows: %d\n", count);
}